document_stream.h

#ifndef SIMDJSON_GENERIC_ONDEMAND_DOCUMENT_STREAM_H
 
#ifndef SIMDJSON_CONDITIONAL_INCLUDE
#define SIMDJSON_GENERIC_ONDEMAND_DOCUMENT_STREAM_H
#include "simdjson/generic/ondemand/base.h"
#include "simdjson/generic/implementation_simdjson_result_base.h"
#include "simdjson/generic/ondemand/document.h"
#include "simdjson/generic/ondemand/parser.h"
#endif // SIMDJSON_CONDITIONAL_INCLUDE
 
#ifdef SIMDJSON_THREADS_ENABLED
#include <thread>
#include <mutex>
#include <condition_variable>
#endif
 
namespace simdjson {
namespace SIMDJSON_IMPLEMENTATION {
namespace ondemand {
 
#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, parser * stage1, size_t next_batch_start);
  void finish();
 
private:
 
  void stop_thread();
 
  std::thread thread{};
  ondemand::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{};
 
  friend class document_stream;
};
#endif  // SIMDJSON_THREADS_ENABLED
 
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<document>;
    using reference  = value_type;
 
    using difference_type   = std::ptrdiff_t;
 
    using iterator_category = std::input_iterator_tag;
 
    simdjson_inline iterator() noexcept;
    simdjson_inline simdjson_result<ondemand::document_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;
 
     inline error_code error() const noexcept;
 
  private:
    simdjson_inline iterator(document_stream *s, bool finished) noexcept;
    document_stream* stream;
    bool finished;
 
    friend class document;
    friend class document_stream;
    friend class json_iterator;
  };
 
  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(
    ondemand::parser &parser,
    const uint8_t *buf,
    size_t len,
    size_t batch_size,
    bool allow_comma_separated
  ) noexcept;
 
  inline void start() noexcept;
 
  inline void next() noexcept;
 
  inline void next_document() noexcept;
 
  inline size_t next_batch_start() const noexcept;
 
  inline error_code run_stage1(ondemand::parser &p, size_t batch_start) noexcept;
 
  // Fields
  ondemand::parser *parser;
  const uint8_t *buf;
  size_t len;
  size_t batch_size;
  bool allow_comma_separated;
  document doc{};
  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};
  std::unique_ptr<stage1_worker> worker{new(std::nothrow) stage1_worker()};
  ondemand::parser stage1_thread_parser{};
 
  friend struct stage1_worker;
  #endif // SIMDJSON_THREADS_ENABLED
 
  friend class parser;
  friend class document;
  friend class json_iterator;
  friend struct simdjson_result<ondemand::document_stream>;
  friend struct internal::simdjson_result_base<ondemand::document_stream>;
};  // document_stream
 
} // namespace ondemand
} // namespace SIMDJSON_IMPLEMENTATION
} // namespace simdjson
 
namespace simdjson {
template<>
struct simdjson_result<SIMDJSON_IMPLEMENTATION::ondemand::document_stream> : public SIMDJSON_IMPLEMENTATION::implementation_simdjson_result_base<SIMDJSON_IMPLEMENTATION::ondemand::document_stream> {
public:
  simdjson_inline simdjson_result(SIMDJSON_IMPLEMENTATION::ondemand::document_stream &&value) noexcept; 
  simdjson_inline simdjson_result(error_code error) noexcept; 
  simdjson_inline simdjson_result() noexcept = default;
};
 
} // namespace simdjson
 
#endif // SIMDJSON_GENERIC_ONDEMAND_DOCUMENT_STREAM_H