Function TEST

pj_datastore/tests/plugin_parser_object_write_test.cpp:93–152

}

TEST(ParserObjectWriteHostTest, ParserWritesToBothHostsFromOneParse) {
  // Host setup: one scalar topic + one object topic.
  DataEngine engine;
  auto dataset_or = engine.createDataset(DatasetDescriptor{.source_name = "t", .time_domain_id = 0});
  ASSERT_TRUE(dataset_or.has_value()) << dataset_or.error();
  PJ_data_source_handle_t source_handle{static_cast<uint32_t>(*dataset_or)};

  // Scalar: ensure topic + DatastoreParserWriteHost bound to it.
  DatastoreSourceWriteHost scalar_impl(engine, source_handle);
  auto scalar_view = sdk::SourceWriteHostView{scalar_impl.raw()};
  const auto topic = *scalar_view.ensureTopic("media_topic");
  DatastoreParserWriteHost parser_write_impl(engine, topic);

  // Object: register topic in ObjectStore; bind DatastoreParserObjectWriteHost.
  ObjectStore store;
  DatastoreSourceObjectWriteHost obj_source(store, *dataset_or);
  const auto obj_topic =
      *sdk::SourceObjectWriteHostView{obj_source.raw()}.registerTopic("media_topic", R"({"media_class":"image"})");
  DatastoreParserObjectWriteHost parser_obj_impl(store, obj_topic.id);

  // Build the registry with both services.
  MockRegistryState registry_state;
  const auto scalar_raw = parser_write_impl.raw();
  const auto obj_raw = parser_obj_impl.raw();
  registry_state.services[ParserWriteHostService::kName] = PJ_service_t{scalar_raw.ctx, scalar_raw.vtable};
  registry_state.services[ParserObjectWriteHostService::kName] = PJ_service_t{obj_raw.ctx, obj_raw.vtable};

  static const PJ_service_registry_vtable_t registry_vtable = {
      PJ_PLUGIN_DATA_API_VERSION,
      sizeof(PJ_service_registry_vtable_t),
      mockGetService,
  };
  const PJ_service_registry_t registry_raw{&registry_state, &registry_vtable};

  // Bind the parser through the SDK.
  MediaParser parser;
  ASSERT_TRUE(parser.bind(sdk::ServiceRegistry{registry_raw}).has_value());

  // parse() one message: seq=7, payload=[0xAA 0xBB 0xCC].
  std::vector<uint8_t> payload(sizeof(uint64_t) + 3);
  uint64_t seq = 7;
  std::memcpy(payload.data(), &seq, sizeof(uint64_t));
  payload[sizeof(uint64_t) + 0] = 0xAA;
  payload[sizeof(uint64_t) + 1] = 0xBB;
  payload[sizeof(uint64_t) + 2] = 0xCC;

  ASSERT_TRUE(parser.parse(100, Span<const uint8_t>(payload.data(), payload.size())).has_value());

  // Object-store side: bytes landed.
  auto resolved = store.latestAt(ObjectTopicId{obj_topic.id}, 100);
  ASSERT_TRUE(resolved.has_value());
  ASSERT_NE(resolved->payload.anchor, nullptr);
  const std::vector<uint8_t> expected{0xAA, 0xBB, 0xCC};
  EXPECT_TRUE(
      std::equal(resolved->payload.bytes.begin(), resolved->payload.bytes.end(), expected.begin(), expected.end()));

  // (Scalar side requires flushing + a read path; Phase-3 scope is proving
  // both hosts were resolved and invoked. Scalar writes go into DataEngine