v1.0.0-rc.36

2025-05-05 19:32:22 +02:00
parent 759243d1bf
commit 040cf08386
333 changed files with 22724 additions and 15251 deletions
--- a/image_puller/ZMQImagePuller.cpp
+++ b/image_puller/ZMQImagePuller.cpp
@@ -0,0 +1,121 @@
+// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "ZMQImagePuller.h"
+#include "../frame_serialize/CBORStream2Serializer.h"
+
+ZMQImagePuller::ZMQImagePuller(const std::string &in_addr,
+                               const std::string &repub_address,
+    const std::optional<int32_t> &rcv_watermark,
+    const std::optional<int32_t> &repub_watermark) :
+        socket (ZMQSocketType::Pull), addr(in_addr) {
+    puller_thread = std::thread(&ZMQImagePuller::PullerThread, this);
+    cbor_thread = std::thread(&ZMQImagePuller::CBORThread, this);
+
+    socket.ReceiveWaterMark(rcv_watermark.value_or(default_receive_watermark));
+    socket.ReceiveTimeout(ReceiveTimeout);
+    socket.Connect(addr);
+
+    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);
+        repub_thread = std::thread(&ZMQImagePuller::RepubThread, this);
+    }
+}
+
+ZMQImagePuller::~ZMQImagePuller() {
+    Disconnect();
+}
+
+void ZMQImagePuller::Disconnect() {
+    disconnect = 1;
+
+    if (puller_thread.joinable())
+        puller_thread.join();
+    if (cbor_thread.joinable())
+        cbor_thread.join();
+    if (repub_thread.joinable())
+        repub_thread.join();
+
+    if (!addr.empty())
+        socket.Disconnect(addr);
+
+    addr = "";
+}
+
+void ZMQImagePuller::PullerThread() {
+    while (true) {
+        ImagePullerOutput ret;
+        ret.msg = std::make_shared<ZMQMessage>();
+        bool received = false;
+        while (!received) {
+            if (disconnect) {
+                cbor_fifo.PutBlocking(ImagePullerOutput{});
+                return;
+            }
+            try {
+                received = socket.Receive(*ret.msg, false);
+                if (!received)
+                    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            } catch (const JFJochException &e) {
+                logger.ErrorException(e);
+            }
+        }
+        cbor_fifo.PutBlocking(ret);
+    }
+}
+
+void ZMQImagePuller::CBORThread() {
+    auto ret = cbor_fifo.GetBlocking();
+
+    while (ret.msg) {
+        try {
+            ret.cbor = CBORStream2Deserialize(ret.msg->data(), ret.msg->size());
+            outside_fifo.PutBlocking(ret);
+            if (repub_socket) {
+                if ((ret.cbor->msg_type == CBORImageType::START)
+                    || (ret.cbor->msg_type == CBORImageType::END))
+                    repub_fifo.PutBlocking(ret);
+                else
+                    repub_fifo.Put(ret);
+            }
+        } catch (const JFJochException &e) {
+            logger.ErrorException(e);
+        }
+        ret = cbor_fifo.GetBlocking();
+    }
+    if (repub_socket)
+        repub_fifo.PutBlocking(ret);
+    outside_fifo.PutBlocking(ret);
+}
+
+void ZMQImagePuller::RepubThread() {
+    auto ret = repub_fifo.GetBlocking();
+    bool repub_active = false;
+
+    while (ret.msg) {
+        try {
+            if (ret.cbor->msg_type == CBORImageType::START) {
+                // Start message needs to be cleaned when running republish
+                StartMessage msg = ret.cbor->start_message.value();
+                msg.writer_notification_zmq_addr = "";
+                std::vector<uint8_t> serialization_buffer(256*1024*1024);
+                CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
+                serializer.SerializeSequenceStart(msg);
+                repub_active = repub_socket->Send(serialization_buffer.data(), serializer.GetBufferSize(), true);
+                if (repub_active)
+                    logger.Info("Republish active");
+            } else {
+                if (repub_active)
+                    repub_socket->Send(ret.msg->data(), ret.msg->size(), true);
+            }
+        } catch (const JFJochException &e) {
+            logger.ErrorException(e);
+        }
+        ret = repub_fifo.GetBlocking();
+    }
+    if (repub_active)
+        logger.Info("Republish finished");
+}