Function sourceEnsureTopic

pj_datastore/src/plugin_data_host.cpp:1267–1279

}

bool sourceEnsureTopic(void* ctx, PJ_string_view_t topic_name, TopicHandle* out_topic, PJ_error_t* out_error) noexcept {
  return guardHostCallback(out_error, [&] {
    auto* impl = static_cast<DatastoreSourceWriteHostState*>(ctx);
    // Pin the current WriteCore for the whole call so a concurrent setTarget()
    // swap can't destroy it under us (and so the op + its error read the SAME core).
    auto core = impl->core.load(std::memory_order_acquire);
    if (!core->ensureTopic(impl->source, toStringView(topic_name), out_topic)) {
      propagateError(out_error, core->lastError());
      return false;
    }
    return true;
  });
}

bool sourceEnsureField(
    void* ctx, TopicHandle topic, PJ_string_view_t field_name, PJ_primitive_type_t type, FieldHandle* out_field,