From 4f79e12b9248b735247e33c14cec2e72693d6c5f Mon Sep 17 00:00:00 2001
From: leonarski_f <filip.leonarski@psi.ch>
Date: Fri, 13 Mar 2026 16:59:21 +0100
Subject: [PATCH] HDF5FilePusher: Add republish socket

---
 image_pusher/HDF5FilePusher.cpp | 59 ++++++++++++++++++++++++++++++++-
 image_pusher/HDF5FilePusher.h   | 12 ++++++-
 2 files changed, 69 insertions(+), 2 deletions(-)

diff --git a/image_pusher/HDF5FilePusher.cpp b/image_pusher/HDF5FilePusher.cpp
index 2797971d..d6c00c74 100644
--- a/image_pusher/HDF5FilePusher.cpp
+++ b/image_pusher/HDF5FilePusher.cpp
@@ -3,7 +3,17 @@
 
 #include "HDF5FilePusher.h"
 #include "../frame_serialize/CBORStream2Deserializer.h"
+#include "../frame_serialize/CBORStream2Serializer.h"
 
+HDF5FilePusher::HDF5FilePusher(const std::string &repub_address,
+                               const std::optional<int32_t> &repub_watermark) {
+    if (!repub_address.empty()) {
+        repub_socket = std::make_unique<ZMQSocket>(ZMQSocketType::Push);
+        repub_socket->SendWaterMark(repub_watermark.value_or(default_repub_watermark));
+        repub_socket->SendTimeout(RepubTimeout);
+        repub_socket->Bind(repub_address);
+    }
+}
 
 void HDF5FilePusher::StartDataCollection(StartMessage &message) {
     if (writer)
@@ -11,6 +21,27 @@ void HDF5FilePusher::StartDataCollection(StartMessage &message) {
     writer = std::make_unique<FileWriter>(message);
     writer_future = std::async(std::launch::async, &HDF5FilePusher::WriterThread, this);
     images_written = 0;
+
+    if (repub_socket) {
+        try {
+            StartMessage repub_message = message;
+            repub_message.writer_notification_zmq_addr = "";
+
+            size_t approx_size = 1024 * 1024;
+            for (const auto &[key, value] : repub_message.pixel_mask)
+                approx_size += value.size() * sizeof(uint32_t);
+
+            std::vector<uint8_t> serialization_buffer(approx_size);
+            CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
+            serializer.SerializeSequenceStart(repub_message);
+
+            repub_active = repub_socket->Send(serialization_buffer.data(), serializer.GetBufferSize(), true);
+            if (repub_active)
+                logger.Info("Republish active");
+        } catch (const JFJochException &e) {
+            logger.ErrorException(e);
+        }
+    }
 }
 
 bool HDF5FilePusher::EndDataCollection(const EndMessage &message) {
@@ -26,6 +57,24 @@ bool HDF5FilePusher::EndDataCollection(const EndMessage &message) {
     writer->Finalize();
     writer.reset();
 
+    if (repub_socket) {
+        try {
+            size_t approx_size = 1024 * 1024;
+            std::vector<uint8_t> serialization_buffer(approx_size);
+            CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
+            serializer.SerializeSequenceEnd(message);
+
+            if (repub_active)
+                repub_socket->Send(serialization_buffer.data(), serializer.GetBufferSize(), true);
+        } catch (const JFJochException &e) {
+            logger.ErrorException(e);
+        }
+
+        if (repub_active)
+            logger.Info("Republish finished");
+        repub_active = false;
+    }
+
     return true;
 }
 
@@ -37,6 +86,14 @@ bool HDF5FilePusher::SendImage(const uint8_t *image_data, size_t image_size, int
     if (deserialized->data_message) {
         writer->Write(*deserialized->data_message);
         images_written++;
+
+        if (repub_socket && repub_active) {
+            try {
+                repub_socket->Send(image_data, image_size, false);
+            } catch (const JFJochException &e) {
+                logger.ErrorException(e);
+            }
+        }
     } else
         throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
                               "HDF5FilePusher::SendImage accepts only data image");
@@ -79,4 +136,4 @@ std::optional<uint64_t> HDF5FilePusher::GetImagesWritten() const {
 
 size_t HDF5FilePusher::GetConnectedWriters() const {
     return 1;
-}
+}
\ No newline at end of file
diff --git a/image_pusher/HDF5FilePusher.h b/image_pusher/HDF5FilePusher.h
index 25f5603a..3d65c0a7 100644
--- a/image_pusher/HDF5FilePusher.h
+++ b/image_pusher/HDF5FilePusher.h
@@ -10,6 +10,8 @@
 #include "ImagePusher.h"
 #include "../writer/FileWriter.h"
 #include "../common/ThreadSafeFIFO.h"
+#include "../common/ZMQWrappers.h"
+#include "../common/Logger.h"
 
 class HDF5FilePusher : public ImagePusher {
     std::unique_ptr<FileWriter> writer;
@@ -19,7 +21,16 @@ class HDF5FilePusher : public ImagePusher {
     void WriterThread();
 
     std::atomic<uint64_t> images_written = 0;
+
+    static constexpr uint32_t default_repub_watermark = 220;
+    static constexpr auto RepubTimeout = std::chrono::milliseconds(100);
+
+    std::unique_ptr<ZMQSocket> repub_socket;
+    bool repub_active = false;
+    Logger logger{"HDF5FilePusher"};
 public:
+    explicit HDF5FilePusher(const std::string &repub_address = "",
+                            const std::optional<int32_t> &repub_watermark = {});
     // Thread safety: StartDataCollection, EndDataCollection and SendCalibration must run poorly in serial context
     // SendImage can be executed in parallel
     void StartDataCollection(StartMessage &message) override;
@@ -35,5 +46,4 @@ public:
     ImagePusherType GetType() const override { return ImagePusherType::HDF5; }
 };
 
-
 #endif //JFJOCH_HDF5FILEPUSHER_H