HDF5FilePusher: Add republish socket

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;
-}
+}