Method cbEnsureParserBinding

pj_runtime/src/DataSourceRuntimeHost.cpp:421–589

}

bool DataSourceRuntimeHost::cbEnsureParserBinding(
    void* ctx, const PJ_parser_binding_request_t* request, PJ_parser_binding_handle_t* out,
    PJ_error_t* out_error) noexcept {
  auto* self = static_cast<DataSourceRuntimeHost*>(ctx);
  try {
    const std::string_view encoding(request->parser_encoding.data, request->parser_encoding.size);
    const std::string_view topic_name(request->topic_name.data, request->topic_name.size);
    const std::string_view type_name(request->type_name.data, request->type_name.size);

    const LoadedMessageParser* parser_entry =
        self->catalog_.findParserByEncoding(QString::fromUtf8(encoding.data(), static_cast<int>(encoding.size())));
    if (parser_entry == nullptr) {
      return self->fail(out_error, ("no parser found for encoding '" + std::string(encoding) + "'").c_str());
    }

    auto parser = std::make_unique<MessageParserHandle>(parser_entry->library.createHandle());
    if (!parser->valid()) {
      return self->fail(out_error, ("failed to create parser instance for '" + std::string(encoding) + "'").c_str());
    }

    auto topic_or = self->engine_.createTopic(self->dataset_id_, TopicDescriptor{.name = std::string(topic_name)});
    if (!topic_or.has_value()) {
      return self->fail(
          out_error, ("failed to create topic '" + std::string(topic_name) + "': " + topic_or.error()).c_str());
    }
    const PJ_topic_handle_t topic_handle{static_cast<uint32_t>(*topic_or)};

    // Lockstep-mirror into the secondary engine with the SAME TopicId. The two
    // engines' TopicId counters drift whenever the primary gets topics the
    // secondary doesn't (e.g. a file loaded between streams), so we force the
    // secondary topic id to match the primary's instead of relying on the
    // counters staying in step. A later push uses one id against whichever
    // engine is the active target, so the ids MUST match.
    if (self->secondary_data_engine_ != nullptr) {
      auto mirrored = self->secondary_data_engine_->createTopic(
          self->dataset_id_, TopicDescriptor{.name = std::string(topic_name)}, *topic_or);
      if (!mirrored.has_value()) {
        return self->fail(
            out_error,
            ("failed to mirror topic '" + std::string(topic_name) + "' into secondary engine: " + mirrored.error())
                .c_str());
      }
    }

    auto write_host = std::make_unique<DatastoreParserWriteHost>(self->engine_, topic_handle);
    // Same lockstep wiring as the source-level write host (see the runtime
    // host constructor). Closes the latent FieldHandle-stale bug for parser
    // plugins that cache handles across messages (parser_protobuf et al.):
    // every ensureField inside parserAppendRecord/appendBoundRecord is now
    // mirrored to the secondary engine via DataEngine::createTopicField with
    // the same FieldId, so the cached handle resolves after a pause/resume
    // target swap.
    write_host->setSecondaryEngine(self->secondary_data_engine_);

    // Build the service registry the parser binds against. The builder must
    // outlive bind() because the plugin may hold a view into it; we move it
    // into the ParserBinding so its lifetime matches the parser's.
    auto registry_builder = std::make_unique<ServiceRegistryBuilder>();