Method DataSourceRuntimeHost

pj_runtime/src/DataSourceRuntimeHost.cpp:237–261

// ---------------------------------------------------------------------------

DataSourceRuntimeHost::DataSourceRuntimeHost(
    DataEngine& engine, ExtensionCatalogService& catalog, DatasetId dataset_id, PJ_data_source_handle_t source_handle,
    ObjectStore& object_store, std::string source_id, ObjectTopicParserRegistrar parser_registrar,
    ObjectStore* secondary_object_store, DataEngine* secondary_data_engine, std::shared_ptr<void> library_keepalive)
    : engine_(engine),
      catalog_(catalog),
      object_store_(object_store),
      secondary_object_store_(secondary_object_store),
      secondary_data_engine_(secondary_data_engine),
      source_id_(std::move(source_id)),
      object_topic_parser_registrar_(std::move(parser_registrar)),
      dataset_id_(dataset_id),
      source_write_host_(engine, source_handle),
      source_object_write_host_(object_store, dataset_id),
      lazy_fetch_mutex_(std::make_shared<std::mutex>()),
      library_keepalive_(std::move(library_keepalive)) {
  // Wire the source-level write host with the secondary engine for the
  // streaming pause/resume two-engine lockstep. Without this, a plugin that
  // caches TopicHandle/FieldHandle on start() (e.g. data_stream_dummy) sees
  // them go stale after the first pause — the secondary engine has no
  // matching ids — and the worker dies on the first post-pause write.
  // Mirroring happens inside DatastoreSourceWriteHost via DataEngine's new
  // createTopic(requested_id) + createTopicField(requested_id) primitives.
  source_write_host_.setSecondaryEngine(secondary_data_engine_);
}

DataSourceRuntimeHost::~DataSourceRuntimeHost() = default;