Merge plotjuggler/main into internal_main (sync 2026-06-01: push_message API consolidation #110)

Author: pabloinigoblasco

pj_base/CMakeLists.txt

@@ -88,7 +88,7 @@ if(PJ_BUILD_TESTS)
     tests/image_annotations_codec_test.cpp
     tests/image_annotations_decoder_test.cpp
     tests/media_metadata_test.cpp
-    tests/push_message_v2_test.cpp
+    tests/push_message_test.cpp
     tests/image_codec_test.cpp
     tests/depth_image_codec_test.cpp
     tests/point_cloud_codec_test.cpp

pj_base/include/pj_base/data_source_protocol.h

@@ -133,7 +133,7 @@ enum {
   PJ_DATA_SOURCE_CAPABILITY_HAS_DIALOG = 1ull << 5,        /**< Plugin provides a configuration dialog. */
 };
 
-/** Opaque handle returned by ensure_parser_binding, used with push_raw_message. */
+/** Opaque handle returned by ensure_parser_binding, used with push_message. */
 typedef struct {
   uint32_t id;
 } PJ_parser_binding_handle_t;
@@ -260,22 +260,6 @@ typedef struct PJ_data_source_runtime_host_vtable_t {
       void* ctx, const PJ_parser_binding_request_t* request, PJ_parser_binding_handle_t* out_handle,
       PJ_error_t* out_error) PJ_NOEXCEPT;
 
-  /**
-   * [stream-thread] Push a raw message payload for host-side parsing.
-   * @p handle must have been obtained from ensure_parser_binding.
-   * @p host_timestamp_ns is nanoseconds since the Unix epoch
-   * (1970-01-01T00:00:00Z). Returns false + error on failure.
-   *
-   * Eager-only push: the host parses immediately and the bytes are not
-   * retained for later replay. Plugins that need lazy materialization or
-   * ObjectIngestPolicy dispatch should use push_message_v2 instead. This
-   * slot remains for sources that fan-out raw bytes without an associated
-   * FetchMessageData callable (streaming or eager-only consumers).
-   */
-  bool (*push_raw_message)(
-      void* ctx, PJ_parser_binding_handle_t handle, int64_t host_timestamp_ns, PJ_bytes_view_t payload,
-      PJ_error_t* out_error) PJ_NOEXCEPT;
-
   /**
    * [main-thread] Display a modal message box to the user and wait for
    * their response. BLOCKS until the user closes the dialog. The host is
@@ -340,7 +324,7 @@ typedef struct PJ_data_source_runtime_host_vtable_t {
    * ObjectStore push failed, etc.). On failure the host still calls
    * `release` so the plugin's ctx leaks no resources.
    */
-  bool (*push_message_v2)(
+  bool (*push_message)(
       void* ctx, PJ_parser_binding_handle_t handle, int64_t host_timestamp_ns,
       PJ_message_data_fetcher_t fetch_message_data, PJ_error_t* out_error) PJ_NOEXCEPT;
 } PJ_data_source_runtime_host_vtable_t;

pj_base/include/pj_base/sdk/data_source_host_views.hpp

@@ -209,19 +209,6 @@ class DataSourceRuntimeHostView {
     return handle;
   }
 
-  /// Push a raw message for host-side parsing via a previously obtained binding handle.
-  [[nodiscard]] Status pushRawMessage(
-      ParserBindingHandle handle, Timestamp host_timestamp_ns, Span<const uint8_t> payload) const {
-    if (!valid() || host_.vtable->push_raw_message == nullptr) {
-      return unexpected("runtime host is not bound");
-    }
-    PJ_error_t err{};
-    if (!host_.vtable->push_raw_message(host_.ctx, handle, host_timestamp_ns, sdk::toAbiBytes(payload), &err)) {
-      return unexpected(errorToString(err));
-    }
-    return okStatus();
-  }
-
   /// Push a message via a deferred FetchMessageData callable. The DataSource
   /// hands the host a callable that produces the payload bytes when invoked.
   /// The host applies the active ObjectIngestPolicy (resolved via the
@@ -245,12 +232,11 @@ class DataSourceRuntimeHostView {
   ///     heap-relocated and used as its own anchor; bytes survive across
   ///     the C ABI boundary at the cost of one alloc-and-move.
   ///
-  /// The host MUST advertise the push_message_v2 tail slot. We wrap the
+  /// The host MUST advertise the push_message tail slot. We wrap the
   /// closure into a PJ_message_data_fetcher_t and hand it over verbatim; the
   /// host applies ObjectIngestPolicy and decides when (and whether) to
   /// invoke it. There is no legacy fallback: a host that doesn't expose
-  /// the slot returns an explicit error here rather than silently
-  /// degrading to a kEager push_raw_message.
+  /// the slot returns an explicit error here.
   template <typename FetchMessageData>
   [[nodiscard]] Status pushMessage(
       ParserBindingHandle handle, Timestamp host_timestamp_ns, FetchMessageData&& fetch_message_data) const {
@@ -264,8 +250,8 @@ class DataSourceRuntimeHostView {
     if (!valid()) {
       return unexpected(std::string("runtime host is not bound"));
     }
-    if (!PJ_HAS_TAIL_SLOT(PJ_data_source_runtime_host_vtable_t, host_.vtable, push_message_v2)) {
-      return unexpected(std::string("runtime host does not expose push_message_v2"));
+    if (!PJ_HAS_TAIL_SLOT(PJ_data_source_runtime_host_vtable_t, host_.vtable, push_message)) {
+      return unexpected(std::string("runtime host does not expose push_message"));
     }
 
     auto* ctx = new FetchMessageDataT(std::forward<FetchMessageData>(fetch_message_data));
@@ -308,7 +294,7 @@ class DataSourceRuntimeHostView {
     };
 
     PJ_error_t err{};
-    if (!host_.vtable->push_message_v2(host_.ctx, handle, host_timestamp_ns, abi_fetch_message_data, &err)) {
+    if (!host_.vtable->push_message(host_.ctx, handle, host_timestamp_ns, abi_fetch_message_data, &err)) {
       return unexpected(errorToString(err));
     }
     return okStatus();

pj_base/include/pj_base/sdk/data_source_patterns.hpp

@@ -136,7 +136,7 @@ class StreamSourceBase : public DataSourcePluginBase {
   /// MUST NOT BLOCK — drain buffered data and return immediately.
   /// Do not call recv(), read(), or any syscall that may wait.
   /// If your source has a receive thread, swap-drain a buffer here.
-  /// Host methods (appendRecord, pushRawMessage) may only be called from this method.
+  /// Host methods (appendRecord, pushMessage) may only be called from this method.
   virtual Status onPoll() = 0;
 
   /// Called from stop(). Close connections, free resources.

pj_base/tests/abi_layout_sentinels_test.cpp

@@ -147,14 +147,11 @@ static_assert(offsetof(PJ_data_source_runtime_host_vtable_t, protocol_version) =
 static_assert(offsetof(PJ_data_source_runtime_host_vtable_t, struct_size) == 4, "v1 prefix pinned");
 static_assert(offsetof(PJ_data_source_runtime_host_vtable_t, report_message) == 8, "v1 first slot pinned");
 static_assert(
-    offsetof(PJ_data_source_runtime_host_vtable_t, push_raw_message) == 72, "v1 push_raw_message slot pinned");
+    offsetof(PJ_data_source_runtime_host_vtable_t, list_available_encodings) == 80,
+    "list_available_encodings slot pinned");
+static_assert(offsetof(PJ_data_source_runtime_host_vtable_t, push_message) == 88, "push_message tail slot pinned");
 static_assert(
-    offsetof(PJ_data_source_runtime_host_vtable_t, list_available_encodings) == 88,
-    "v1 list_available_encodings slot pinned");
-static_assert(
-    offsetof(PJ_data_source_runtime_host_vtable_t, push_message_v2) == 96, "v1 push_message_v2 tail slot pinned");
-static_assert(
-    sizeof(PJ_data_source_runtime_host_vtable_t) == 104, "Runtime host vtable size (update deliberately on append)");
+    sizeof(PJ_data_source_runtime_host_vtable_t) == 96, "Runtime host vtable size (update deliberately on append)");
 
 // --- Write-host vtables (ABI-APPENDABLE within v4) --------------------------
 static_assert(offsetof(PJ_source_write_host_vtable_t, abi_version) == 0, "source write host prefix pinned");

pj_base/tests/push_message_v2_test.cpp pj_base/tests/push_message_test.cpppj_base/tests/push_message_v2_test.cpp renamed to pj_base/tests/push_message_test.cpp

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 // Tests for the SDK template `DataSourceRuntimeHostView::pushMessage` and
-// its delegation to the C ABI slot `push_message_v2`. We exercise:
+// its delegation to the C ABI slot `push_message`. We exercise:
 //
 //   1. Vector closure → the captured FetchMessageData callable yields the
 //      same bytes when invoked from the host.
@@ -11,7 +11,7 @@
 //   3. Multiple invocations are idempotent (same bytes each time).
 //   4. The heap-held closure context is destroyed exactly once when the
 //      host calls `release`.
-//   5. When the host does not expose push_message_v2 (struct_size short
+//   5. When the host does not expose push_message (struct_size short
 //      or field NULL), pushMessage returns an explicit error rather than
 //      degrading silently.
 
@@ -28,32 +28,30 @@
 
 namespace {
 
-// Captured state from a push_message_v2 invocation.
+// Captured state from a push_message invocation.
 struct CapturedPush {
   PJ_parser_binding_handle_t handle{};
   int64_t timestamp_ns = 0;
   PJ_message_data_fetcher_t fetch_message_data{};
   bool received = false;
 };
 
-// Mock runtime host — exposes a vtable that captures push_message_v2 calls
-// and, alternatively, push_raw_message calls (for the legacy fallback).
+// Mock runtime host — exposes a vtable that captures push_message calls.
 class MockHost {
  public:
   MockHost() {
     vtable_.protocol_version = PJ_DATA_SOURCE_PROTOCOL_VERSION;
     vtable_.struct_size = sizeof(PJ_data_source_runtime_host_vtable_t);
-    vtable_.push_raw_message = &MockHost::pushRawMessageThunk;
-    vtable_.push_message_v2 = &MockHost::pushMessageV2Thunk;
+    vtable_.push_message = &MockHost::pushMessageThunk;
     host_.ctx = this;
     host_.vtable = &vtable_;
   }
 
   // Drop the v2 slot — both clearing the field and shrinking struct_size,
   // matching the runtime scenario where the host predates the addition.
-  void disablePushMessageV2() {
-    vtable_.push_message_v2 = nullptr;
-    vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, push_message_v2);
+  void disablePushMessage() {
+    vtable_.push_message = nullptr;
+    vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, push_message);
   }
 
   PJ::DataSourceRuntimeHostView view() const {
@@ -63,20 +61,9 @@ class MockHost {
   CapturedPush& captured() {
     return captured_;
   }
-  std::vector<uint8_t>& receivedRawBytes() {
-    return raw_bytes_;
-  }
 
  private:
-  static bool pushRawMessageThunk(
-      void* ctx, PJ_parser_binding_handle_t /*handle*/, int64_t /*ts*/, PJ_bytes_view_t payload,
-      PJ_error_t* /*err*/) noexcept {
-    auto* self = static_cast<MockHost*>(ctx);
-    self->raw_bytes_.assign(payload.data, payload.data + payload.size);
-    return true;
-  }
-
-  static bool pushMessageV2Thunk(
+  static bool pushMessageThunk(
       void* ctx, PJ_parser_binding_handle_t handle, int64_t ts, PJ_message_data_fetcher_t fetch_message_data,
       PJ_error_t* /*err*/) noexcept {
     auto* self = static_cast<MockHost*>(ctx);
@@ -90,7 +77,6 @@ class MockHost {
   PJ_data_source_runtime_host_vtable_t vtable_{};
   PJ_data_source_runtime_host_t host_{};
   CapturedPush captured_;
-  std::vector<uint8_t> raw_bytes_;
 };
 
 // Helper: invoke a captured FetchMessageData callable and assert the produced bytes match
@@ -108,9 +94,9 @@ void invokeFetchMessageDataAndExpect(
   }
 }
 
-// ---------- Tests against the new push_message_v2 path ----------
+// ---------- Tests against the new push_message path ----------
 
-TEST(PushMessageV2Test, VectorClosureFlowsThroughSlot) {
+TEST(PushMessageTest, VectorClosureFlowsThroughSlot) {
   MockHost host;
   std::vector<uint8_t> expected{1, 2, 3, 4, 5};
 
@@ -124,7 +110,7 @@ TEST(PushMessageV2Test, VectorClosureFlowsThroughSlot) {
   host.captured().fetch_message_data.release(host.captured().fetch_message_data.ctx);
 }
 
-TEST(PushMessageV2Test, PayloadViewClosureFlowsThroughSlot) {
+TEST(PushMessageTest, PayloadViewClosureFlowsThroughSlot) {
   MockHost host;
   std::vector<uint8_t> expected{10, 20, 30};
   auto owned = std::make_shared<std::vector<uint8_t>>(expected);
@@ -138,7 +124,7 @@ TEST(PushMessageV2Test, PayloadViewClosureFlowsThroughSlot) {
   host.captured().fetch_message_data.release(host.captured().fetch_message_data.ctx);
 }
 
-TEST(PushMessageV2Test, FetchIsIdempotent) {
+TEST(PushMessageTest, FetchIsIdempotent) {
   MockHost host;
   std::vector<uint8_t> expected{0x42, 0x43};
 
@@ -151,7 +137,7 @@ TEST(PushMessageV2Test, FetchIsIdempotent) {
   host.captured().fetch_message_data.release(host.captured().fetch_message_data.ctx);
 }
 
-TEST(PushMessageV2Test, FetchMessageDataCtxReleasedAfterHostCalls) {
+TEST(PushMessageTest, FetchMessageDataCtxReleasedAfterHostCalls) {
   MockHost host;
   auto canary = std::make_shared<int>(42);
   std::weak_ptr<int> witness = canary;
@@ -169,7 +155,7 @@ TEST(PushMessageV2Test, FetchMessageDataCtxReleasedAfterHostCalls) {
   EXPECT_TRUE(witness.expired()) << "after release, the captured shared_ptr should have been the last reference";
 }
 
-TEST(PushMessageV2Test, PayloadAnchorPropagates) {
+TEST(PushMessageTest, PayloadAnchorPropagates) {
   MockHost host;
   auto owned = std::make_shared<std::vector<uint8_t>>(std::vector<uint8_t>{0x99, 0x9A});
   std::weak_ptr<std::vector<uint8_t>> witness = owned;
@@ -204,17 +190,16 @@ TEST(PushMessageV2Test, PayloadAnchorPropagates) {
   EXPECT_TRUE(witness.expired());
 }
 
-// ---------- Host without push_message_v2 returns explicit error ----------
+// ---------- Host without push_message returns explicit error ----------
 
-TEST(PushMessageV2Test, ReturnsErrorWhenSlotMissing) {
+TEST(PushMessageTest, ReturnsErrorWhenSlotMissing) {
   MockHost host;
-  host.disablePushMessageV2();
+  host.disablePushMessage();
 
   std::vector<uint8_t> expected{0xA, 0xB, 0xC};
   auto status = host.view().pushMessage(PJ::ParserBindingHandle{1}, 100, [bytes = expected]() { return bytes; });
-  EXPECT_FALSE(status);  // explicit failure — no silent fallback to push_raw_message
+  EXPECT_FALSE(status);  // explicit failure — no silent fallback
   EXPECT_FALSE(host.captured().received);
-  EXPECT_TRUE(host.receivedRawBytes().empty());
 }
 
 }  // namespace

pj_datastore/docs/USER_GUIDE.md

@@ -200,8 +200,9 @@ auto binding = runtimeHost().ensureParserBinding({
     .parser_config_json = config_json,
 });
 
-// In onPoll(): push incoming messages
-runtimeHost().pushRawMessage(*binding, timestamp_ns, payload_span);
+// In onPoll(): push incoming messages (the host invokes the fetcher per policy)
+runtimeHost().pushMessage(
+    *binding, timestamp_ns, [bytes = payload]() -> std::vector<uint8_t> { return bytes; });
 ```
 
 - `parser_encoding` is the wire format (e.g., `"cdr"`, `"json"`, `"protobuf"`), not the schema format
@@ -372,7 +373,8 @@ class MyStreamer : public PJ::StreamSourceBase {
   PJ::Status onPoll() override {
     // Drain buffered messages (must not block!)
     while (auto msg = dequeue()) {
-      runtimeHost().pushRawMessage(binding_, msg->timestamp_ns, msg->payload);
+      runtimeHost().pushMessage(
+          binding_, msg->timestamp_ns, [bytes = msg->payload]() -> std::vector<uint8_t> { return bytes; });
     }
     return PJ::okStatus();
   }

pj_plugins/docs/ARCHITECTURE.md

@@ -528,7 +528,7 @@ out_array, err)`:
 
 ## Builtin-object pipeline (PR #86)
 
-The v4 DataSource runtime host adds a tail slot `push_message_v2`
+The v4 DataSource runtime host adds a tail slot `push_message`
 (offset 96 in `PJ_data_source_runtime_host_vtable_t`) that takes a
 deferred byte-fetch callable instead of bytes:
 
@@ -539,7 +539,7 @@ typedef struct PJ_message_data_fetcher_t {
   void  (*release)(void* ctx);
 } PJ_message_data_fetcher_t;
 
-bool (*push_message_v2)(
+bool (*push_message)(
     void* ctx, PJ_parser_binding_handle_t handle, int64_t timestamp_ns,
     PJ_message_data_fetcher_t fetch_message_data,
     PJ_error_t* out_error) PJ_NOEXCEPT;

pj_plugins/docs/data-source-guide.md

@@ -313,7 +313,7 @@ class UdpReceiver : public PJ::StreamSourceBase {
 
     // Bind a parser for the configured encoding. The host resolves the
     // parser library, creates a parser instance, binds the write host
-    // and schema, and returns a handle for pushRawMessage().
+    // and schema, and returns a handle for pushMessage().
     binding_ = runtimeHost().ensureParserBinding({
         .topic_name = "udp/data",
         .parser_encoding = encoding_,
@@ -345,9 +345,11 @@ class UdpReceiver : public PJ::StreamSourceBase {
     }
 
     for (const auto& msg : batch) {
-      auto status = runtimeHost().pushRawMessage(
+      // pushMessage takes a deferred fetcher; the host invokes it according to
+      // the active ObjectIngestPolicy (eager for plain curves).
+      auto status = runtimeHost().pushMessage(
           *binding_, msg.timestamp,
-          PJ::Span<const uint8_t>(msg.payload.data(), msg.payload.size()));
+          [bytes = msg.payload]() -> std::vector<uint8_t> { return bytes; });
       if (!status) {
         return PJ::unexpected(status.error());
       }
@@ -415,7 +417,7 @@ changes, only a different config value.
 callback for flushing accumulated data. The host calls it periodically from
 its own thread. For sources with asynchronous I/O (sockets, hardware), the
 plugin must manage its own receive thread and use `onPoll()` as the sync
-point where buffered data is handed off. Host methods (`pushRawMessage`,
+point where buffered data is handed off. Host methods (`pushMessage`,
 `appendRecord`, etc.) must only be called from `onPoll()` / the host's
 thread, never from the background thread.
 
@@ -425,7 +427,7 @@ Key traits of `StreamSourceBase`:
 - `onStart()` opens connections and creates topics or parser bindings.
 - `onPoll()` flushes buffered data into the host — drain what your plugin
   has accumulated and return immediately. Must not block. Host methods
-  (`pushRawMessage`, `appendRecord`, etc.) may only be called from this
+  (`pushMessage`, `appendRecord`, etc.) may only be called from this
   callback.
 - `onStop()` tears down connections. Must be idempotent.
 - `stop()`, `start()`, `poll()`, and `currentState()` are managed by the
@@ -465,7 +467,7 @@ Access via `runtimeHost()`. Use this for lifecycle coordination and diagnostics.
 | `requestStop(terminal_state, reason)` | Ask the host to stop you (self-terminate). |
 | `isStopRequested()` | Check if the host wants you to stop. |
 | `ensureParserBinding(request)` | Bind a parser for delegated ingest (see below). |
-| `pushRawMessage(handle, timestamp, payload)` | Push raw bytes through a parser binding. |
+| `pushMessage(handle, timestamp, fetch_message_data)` | Push a message through a parser binding via a deferred fetcher callable; the host invokes it per the active ObjectIngestPolicy (eager/lazy). |
 
 ## Optional Features
 
@@ -543,8 +545,9 @@ auto binding = runtimeHost().ensureParserBinding({
 });
 if (!binding) { return PJ::unexpected(binding.error()); }
 
-// 2. Push raw payloads — the host parses and stores them
-auto status = runtimeHost().pushRawMessage(*binding, timestamp_ns, payload);
+// 2. Push payloads — the host invokes the fetcher and parses/stores per policy
+auto status = runtimeHost().pushMessage(
+    *binding, timestamp_ns, [bytes = payload]() -> std::vector<uint8_t> { return bytes; });
 ```
 
 The host manages parser instances, caches bindings, and handles schema