jfjoch_tests: Add test for HDF5 pusher republish functionality

tests/JFJochReceiverProcessingTest.cpp

@@ -1401,6 +1401,67 @@ TEST_CASE("JFJochIntegrationTest_HDF5FilePusher", "[JFJochReceiver]") {
     CHECK(!receiver_out.status.cancelled);
 }
 
+TEST_CASE("JFJochIntegrationTest_HDF5FilePusher_republish", "[JFJochReceiver]") {
+    Logger logger(Catch::getResultCapture().getCurrentTestName());
+
+    RegisterHDF5Filter();
+
+    const uint16_t nthreads = 4;
+    const std::string repub_addr = "tcp://127.0.0.1:9132";
+
+    DiffractionExperiment experiment(DetJF4M());
+    experiment.ImagesPerTrigger(5).NumTriggers(1).UseInternalPacketGenerator(true)
+            .FilePrefix("HDF5FilePusherRepublishTest").JungfrauConvPhotonCnt(true)
+            .DetectorDistance_mm(75).BeamY_pxl(1136).BeamX_pxl(1090).IncidentEnergy_keV(12.4);
+
+    PixelMask pixel_mask(experiment);
+
+    AcquisitionDeviceGroup aq_devices;
+    aq_devices.Add(std::make_unique<HLSSimulatedDevice>(0, 64));
+
+    HDF5FilePusher pusher(repub_addr);
+
+    ZMQSocket pull_socket(ZMQSocketType::Pull);
+    pull_socket.ReceiveWaterMark(16);
+    pull_socket.Connect(repub_addr);
+    pull_socket.ReceiveTimeout(std::chrono::seconds(1));
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    JFJochReceiverService service(aq_devices, logger, pusher);
+    service.NumThreads(nthreads);
+    service.Indexing(experiment.GetIndexingSettings());
+
+    service.Start(experiment, pixel_mask, nullptr);
+    auto receiver_out = service.Stop();
+
+    CHECK(receiver_out.efficiency == 1.0);
+    CHECK(receiver_out.status.images_collected == 5);
+    CHECK(receiver_out.status.images_sent == 5);
+    CHECK(!receiver_out.status.cancelled);
+
+    ZMQMessage msg;
+
+    REQUIRE(pull_socket.Receive(msg, true));
+    auto out = CBORStream2Deserialize(msg.data(), msg.size());
+    REQUIRE(out != nullptr);
+    CHECK(out->msg_type == CBORImageType::START);
+
+    for (int i = 0; i < experiment.GetImageNum(); i++) {
+        REQUIRE(pull_socket.Receive(msg, true));
+        out = CBORStream2Deserialize(msg.data(), msg.size());
+        REQUIRE(out != nullptr);
+        CHECK(out->msg_type == CBORImageType::IMAGE);
+    }
+
+    REQUIRE(pull_socket.Receive(msg, true));
+    out = CBORStream2Deserialize(msg.data(), msg.size());
+    REQUIRE(out != nullptr);
+    CHECK(out->msg_type == CBORImageType::END);
+
+    REQUIRE(!pull_socket.Receive(msg, true));
+}
+
 TEST_CASE("JFJochIntegrationTest_HDF5FilePusher_cbf", "[JFJochReceiver]") {
     Logger logger(Catch::getResultCapture().getCurrentTestName());