Method onCreateParserIngest

pj_runtime/src/ToolboxRuntimeHost.cpp:131–178

}  // namespace

bool ToolboxRuntimeHost::onCreateParserIngest(
    void* ctx, uint32_t data_source_id, PJ_data_source_runtime_host_t* out_host, PJ_error_t* out_error) noexcept {
  auto* self = static_cast<ToolboxRuntimeHost*>(ctx);
  if (self == nullptr || out_host == nullptr) {
    return parserIngestFail(out_error, "invalid arguments");
  }
  try {
    if (self->parser_ingest_deps_.catalog == nullptr) {
      return parserIngestFail(out_error, "parser ingest is not configured on this host");
    }
    // Same validation the toolbox write host applies to object topics
    // (plugin_data_host.cpp toolboxRegisterObjectTopic): the id must be a
    // live dataset.
    if (self->engine_.getDataset(static_cast<DatasetId>(data_source_id)) == nullptr) {
      return parserIngestFail(out_error, "data source not found — call createDataSource first");
    }
    std::lock_guard lock(self->parser_ingest_mu_);
    // Construct first, insert only on success: a throwing constructor must not
    // leave a null entry in the map (the release path dereferences entries).
    auto it = self->parser_ingests_.find(data_source_id);
    if (it == self->parser_ingests_.end()) {
      auto host = std::make_unique<DataSourceRuntimeHost>(
          self->engine_, *self->parser_ingest_deps_.catalog, static_cast<DatasetId>(data_source_id),
          PJ_data_source_handle_t{data_source_id}, self->object_store_,
          "toolbox-ingest-" + std::to_string(data_source_id), self->parser_ingest_deps_.register_object_parser,
          // No secondary store/engine (toolbox ingest writes straight into the
          // primary), and no plugin-DSO keepalive — the toolbox isn't a
          // DataSource plugin; its render parsers come from the catalog, which
          // owns their .so lifetime and outlives every ingest context.
          /*secondary_object_store=*/nullptr, /*secondary_data_engine=*/nullptr,
          /*library_keepalive=*/std::shared_ptr<void>{});
      it = self->parser_ingests_.emplace(data_source_id, std::move(host)).first;
    }
    // Remember the dataset for the next notify_data_changed: an ingest context
    // marks a bulk import the host will want to focus playback on.
    const auto ds_id = static_cast<DatasetId>(data_source_id);
    auto& pending = self->pending_ingest_datasets_;
    if (std::find(pending.begin(), pending.end(), ds_id) == pending.end()) {
      pending.push_back(ds_id);
    }
    *out_host = it->second->hostHandle();  // out_host written ONLY on success (documented contract)
    return true;
  } catch (const std::exception& e) {
    return parserIngestFail(out_error, e.what());
  } catch (...) {
    return parserIngestFail(out_error, "unknown exception in create_parser_ingest");
  }
}

bool ToolboxRuntimeHost::onReleaseParserIngest(void* ctx, uint32_t data_source_id, PJ_error_t* out_error) noexcept {
  auto* self = static_cast<ToolboxRuntimeHost*>(ctx);