document_stream.h

#ifndef SIMDJSON_DOCUMENT_STREAM_H
#define SIMDJSON_DOCUMENT_STREAM_H
 
#include "simdjson/dom/base.h"
#include "simdjson/dom/parser.h"
 
#ifdef SIMDJSON_THREADS_ENABLED
#include <thread>
#include <mutex>
#include <condition_variable>
#endif
 
namespace simdjson {
namespace dom {
 
#ifdef SIMDJSON_THREADS_ENABLED
struct stage1_worker {
  stage1_worker() noexcept = default;
  stage1_worker(const stage1_worker&) = delete;
  stage1_worker(stage1_worker&&) = delete;
  stage1_worker operator=(const stage1_worker&) = delete;
  ~stage1_worker();
  void start_thread();
  void run(document_stream * ds, dom::parser * stage1, size_t next_batch_start);
  void finish();
 
private:
 
  void stop_thread();
 
  std::thread thread{};
  dom::parser * stage1_thread_parser{};
  size_t _next_batch_start{};
  document_stream * owner{};
  bool has_work{false};
  bool can_work{true};
 
  std::mutex locking_mutex{};
  std::condition_variable cond_var{};
};
#endif
 
class document_stream {
public:
  simdjson_inline document_stream() noexcept;
  simdjson_inline document_stream(document_stream &&other) noexcept = default;
  simdjson_inline document_stream &operator=(document_stream &&other) noexcept = default;
 
  simdjson_inline ~document_stream() noexcept;
  inline size_t size_in_bytes() const noexcept;
  inline size_t truncated_bytes() const noexcept;
  class iterator {
  public:
    using value_type = simdjson_result<element>;
    using reference  = value_type;
 
    using difference_type   = std::ptrdiff_t;
 
    using iterator_category = std::input_iterator_tag;
 
    simdjson_inline iterator() noexcept;
    simdjson_inline reference operator*() noexcept;
    inline iterator& operator++() noexcept;
    simdjson_inline bool operator!=(const iterator &other) const noexcept;
     simdjson_inline size_t current_index() const noexcept;
     simdjson_inline std::string_view source() const noexcept;
 
  private:
    simdjson_inline iterator(document_stream *s, bool finished) noexcept;
    document_stream* stream;
    bool finished;
    friend class document_stream;
  };
 
  simdjson_inline iterator begin() noexcept;
  simdjson_inline iterator end() noexcept;
 
private:
 
  document_stream &operator=(const document_stream &) = delete; // Disallow copying
  document_stream(const document_stream &other) = delete; // Disallow copying
 
  simdjson_inline document_stream(
    dom::parser &parser,
    const uint8_t *buf,
    size_t len,
    size_t batch_size,
    stream_format format = stream_format::whitespace_delimited
  ) noexcept;
 
  inline void start() noexcept;
 
  inline void next() noexcept;
 
  inline void load_batch() noexcept;
 
  inline size_t next_batch_start() const noexcept;
 
  inline error_code run_stage1(dom::parser &p, size_t batch_start) noexcept;
 
  dom::parser *parser;
  const uint8_t *buf;
  size_t len;
  size_t batch_size;
  stream_format format;
  error_code error;
  size_t batch_start{0};
  size_t doc_index{};
#ifdef SIMDJSON_THREADS_ENABLED
  bool use_thread;
 
  inline void load_from_stage1_thread() noexcept;
 
  inline void start_stage1_thread() noexcept;
 
  inline void finish_stage1_thread() noexcept;
 
  error_code stage1_thread_error{UNINITIALIZED};
  friend struct stage1_worker;
  std::unique_ptr<stage1_worker> worker{new(std::nothrow) stage1_worker()};
  dom::parser stage1_thread_parser{};
#endif // SIMDJSON_THREADS_ENABLED
 
  friend class dom::parser;
  friend struct simdjson_result<dom::document_stream>;
  friend struct internal::simdjson_result_base<dom::document_stream>;
 
}; // class document_stream
 
} // namespace dom
 
template<>
struct simdjson_result<dom::document_stream> : public internal::simdjson_result_base<dom::document_stream> {
public:
  simdjson_inline simdjson_result() noexcept; 
  simdjson_inline simdjson_result(error_code error) noexcept; 
  simdjson_inline simdjson_result(dom::document_stream &&value) noexcept; 
 
#if SIMDJSON_EXCEPTIONS
  simdjson_inline dom::document_stream::iterator begin() noexcept(false);
  simdjson_inline dom::document_stream::iterator end() noexcept(false);
#else // SIMDJSON_EXCEPTIONS
#ifndef SIMDJSON_DISABLE_DEPRECATED_API
  [[deprecated("parse_many() and load_many() may return errors. Use document_stream stream; error = parser.parse_many().get(doc); instead.")]]
  simdjson_inline dom::document_stream::iterator begin() noexcept;
  [[deprecated("parse_many() and load_many() may return errors. Use document_stream stream; error = parser.parse_many().get(doc); instead.")]]
  simdjson_inline dom::document_stream::iterator end() noexcept;
#endif // SIMDJSON_DISABLE_DEPRECATED_API
#endif // SIMDJSON_EXCEPTIONS
}; // struct simdjson_result<dom::document_stream>
 
} // namespace simdjson
 
#endif // SIMDJSON_DOCUMENT_STREAM_H