Merge branch 'revert-b77d16aa' into 'internal_main'

Revert "Merge branch 'refactor/push-message-slot-rename' into 'internal_main'"

See merge request client-projets/p.2026-plotjuggler/plotjuggler_core!193

Author: Vlozano-ibrobotics

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_test.cpp
+    tests/push_message_v2_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_message. */
+/** Opaque handle returned by ensure_parser_binding, used with push_raw_message. */
 typedef struct {
   uint32_t id;
 } PJ_parser_binding_handle_t;
@@ -260,6 +260,22 @@ 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
@@ -324,7 +340,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)(
+  bool (*push_message_v2)(
       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,6 +209,19 @@ 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
@@ -232,11 +245,12 @@ 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 tail slot. We wrap the
+  /// The host MUST advertise the push_message_v2 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.
+  /// the slot returns an explicit error here rather than silently
+  /// degrading to a kEager push_raw_message.
   template <typename FetchMessageData>
   [[nodiscard]] Status pushMessage(
       ParserBindingHandle handle, Timestamp host_timestamp_ns, FetchMessageData&& fetch_message_data) const {
@@ -250,8 +264,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)) {
-      return unexpected(std::string("runtime host does not expose push_message"));
+    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"));
     }
 
     auto* ctx = new FetchMessageDataT(std::forward<FetchMessageData>(fetch_message_data));
@@ -294,7 +308,7 @@ class DataSourceRuntimeHostView {
     };
 
     PJ_error_t err{};
-    if (!host_.vtable->push_message(host_.ctx, handle, host_timestamp_ns, abi_fetch_message_data, &err)) {
+    if (!host_.vtable->push_message_v2(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, pushMessage) may only be called from this method.
+  /// Host methods (appendRecord, pushRawMessage) 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,12 +147,14 @@ 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, list_available_encodings) == 80,
-    "list_available_encodings slot pinned");
+    offsetof(PJ_data_source_runtime_host_vtable_t, push_raw_message) == 72, "v1 push_raw_message slot pinned");
 static_assert(
-    offsetof(PJ_data_source_runtime_host_vtable_t, push_message) == 88, "push_message tail slot pinned");
+    offsetof(PJ_data_source_runtime_host_vtable_t, list_available_encodings) == 88,
+    "v1 list_available_encodings slot pinned");
 static_assert(
-    sizeof(PJ_data_source_runtime_host_vtable_t) == 96, "Runtime host vtable size (update deliberately on append)");
+    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)");
 
 // --- 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_test.cpp pj_base/tests/push_message_v2_test.cpppj_base/tests/push_message_test.cpp renamed to pj_base/tests/push_message_v2_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`. We exercise:
+// its delegation to the C ABI slot `push_message_v2`. 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 (struct_size short
+//   5. When the host does not expose push_message_v2 (struct_size short
 //      or field NULL), pushMessage returns an explicit error rather than
 //      degrading silently.
 
@@ -28,30 +28,32 @@
 
 namespace {
 
-// Captured state from a push_message invocation.
+// Captured state from a push_message_v2 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 calls.
+// Mock runtime host — exposes a vtable that captures push_message_v2 calls
+// and, alternatively, push_raw_message calls (for the legacy fallback).
 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_message = &MockHost::pushMessageThunk;
+    vtable_.push_raw_message = &MockHost::pushRawMessageThunk;
+    vtable_.push_message_v2 = &MockHost::pushMessageV2Thunk;
     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 disablePushMessage() {
-    vtable_.push_message = nullptr;
-    vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, push_message);
+  void disablePushMessageV2() {
+    vtable_.push_message_v2 = nullptr;
+    vtable_.struct_size = offsetof(PJ_data_source_runtime_host_vtable_t, push_message_v2);
   }
 
   PJ::DataSourceRuntimeHostView view() const {
@@ -61,9 +63,20 @@ class MockHost {
   CapturedPush& captured() {
     return captured_;
   }
+  std::vector<uint8_t>& receivedRawBytes() {
+    return raw_bytes_;
+  }
 
  private:
-  static bool pushMessageThunk(
+  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(
       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);
@@ -77,6 +90,7 @@ 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
@@ -94,9 +108,9 @@ void invokeFetchMessageDataAndExpect(
   }
 }
 
-// ---------- Tests against the new push_message path ----------
+// ---------- Tests against the new push_message_v2 path ----------
 
-TEST(PushMessageTest, VectorClosureFlowsThroughSlot) {
+TEST(PushMessageV2Test, VectorClosureFlowsThroughSlot) {
   MockHost host;
   std::vector<uint8_t> expected{1, 2, 3, 4, 5};
 
@@ -110,7 +124,7 @@ TEST(PushMessageTest, VectorClosureFlowsThroughSlot) {
   host.captured().fetch_message_data.release(host.captured().fetch_message_data.ctx);
 }
 
-TEST(PushMessageTest, PayloadViewClosureFlowsThroughSlot) {
+TEST(PushMessageV2Test, PayloadViewClosureFlowsThroughSlot) {
   MockHost host;
   std::vector<uint8_t> expected{10, 20, 30};
   auto owned = std::make_shared<std::vector<uint8_t>>(expected);
@@ -124,7 +138,7 @@ TEST(PushMessageTest, PayloadViewClosureFlowsThroughSlot) {
   host.captured().fetch_message_data.release(host.captured().fetch_message_data.ctx);
 }
 
-TEST(PushMessageTest, FetchIsIdempotent) {
+TEST(PushMessageV2Test, FetchIsIdempotent) {
   MockHost host;
   std::vector<uint8_t> expected{0x42, 0x43};
 
@@ -137,7 +151,7 @@ TEST(PushMessageTest, FetchIsIdempotent) {
   host.captured().fetch_message_data.release(host.captured().fetch_message_data.ctx);
 }
 
-TEST(PushMessageTest, FetchMessageDataCtxReleasedAfterHostCalls) {
+TEST(PushMessageV2Test, FetchMessageDataCtxReleasedAfterHostCalls) {
   MockHost host;
   auto canary = std::make_shared<int>(42);
   std::weak_ptr<int> witness = canary;
@@ -155,7 +169,7 @@ TEST(PushMessageTest, FetchMessageDataCtxReleasedAfterHostCalls) {
   EXPECT_TRUE(witness.expired()) << "after release, the captured shared_ptr should have been the last reference";
 }
 
-TEST(PushMessageTest, PayloadAnchorPropagates) {
+TEST(PushMessageV2Test, 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;
@@ -190,16 +204,17 @@ TEST(PushMessageTest, PayloadAnchorPropagates) {
   EXPECT_TRUE(witness.expired());
 }
 
-// ---------- Host without push_message returns explicit error ----------
+// ---------- Host without push_message_v2 returns explicit error ----------
 
-TEST(PushMessageTest, ReturnsErrorWhenSlotMissing) {
+TEST(PushMessageV2Test, ReturnsErrorWhenSlotMissing) {
   MockHost host;
-  host.disablePushMessage();
+  host.disablePushMessageV2();
 
   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
+  EXPECT_FALSE(status);  // explicit failure — no silent fallback to push_raw_message
   EXPECT_FALSE(host.captured().received);
+  EXPECT_TRUE(host.receivedRawBytes().empty());
 }
 
 }  // namespace

pj_datastore/docs/USER_GUIDE.md

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

pj_plugins/docs/ARCHITECTURE.md

@@ -519,7 +519,7 @@ out_array, err)`:
 
 ## Builtin-object pipeline (PR #86)
 
-The v4 DataSource runtime host adds a tail slot `push_message`
+The v4 DataSource runtime host adds a tail slot `push_message_v2`
 (offset 96 in `PJ_data_source_runtime_host_vtable_t`) that takes a
 deferred byte-fetch callable instead of bytes:
 
@@ -530,7 +530,7 @@ typedef struct PJ_message_data_fetcher_t {
   void  (*release)(void* ctx);
 } PJ_message_data_fetcher_t;
 
-bool (*push_message)(
+bool (*push_message_v2)(
     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 pushMessage().
+    // and schema, and returns a handle for pushRawMessage().
     binding_ = runtimeHost().ensureParserBinding({
         .topic_name = "udp/data",
         .parser_encoding = encoding_,
@@ -345,11 +345,9 @@ class UdpReceiver : public PJ::StreamSourceBase {
     }
 
     for (const auto& msg : batch) {
-      // pushMessage takes a deferred fetcher; the host invokes it according to
-      // the active ObjectIngestPolicy (eager for plain curves).
-      auto status = runtimeHost().pushMessage(
+      auto status = runtimeHost().pushRawMessage(
           *binding_, msg.timestamp,
-          [bytes = msg.payload]() -> std::vector<uint8_t> { return bytes; });
+          PJ::Span<const uint8_t>(msg.payload.data(), msg.payload.size()));
       if (!status) {
         return PJ::unexpected(status.error());
       }
@@ -417,7 +415,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 (`pushMessage`,
+point where buffered data is handed off. Host methods (`pushRawMessage`,
 `appendRecord`, etc.) must only be called from `onPoll()` / the host's
 thread, never from the background thread.
 
@@ -427,7 +425,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
-  (`pushMessage`, `appendRecord`, etc.) may only be called from this
+  (`pushRawMessage`, `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
@@ -467,7 +465,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). |
-| `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). |
+| `pushRawMessage(handle, timestamp, payload)` | Push raw bytes through a parser binding. |
 
 ## Optional Features
 
@@ -545,9 +543,8 @@ auto binding = runtimeHost().ensureParserBinding({
 });
 if (!binding) { return PJ::unexpected(binding.error()); }
 
-// 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; });
+// 2. Push raw payloads — the host parses and stores them
+auto status = runtimeHost().pushRawMessage(*binding, timestamp_ns, payload);
 ```
 
 The host manages parser instances, caches bindings, and handles schema