zmq sender and receiver examples

network_io/include/aare/ZmqHeader.hpp

@@ -0,0 +1,106 @@
+#include "aare/utils/logger.hpp"
+#include "simdjson.h"
+#include <array>
+#include <cstdint>
+#include <map>
+#include <string>
+namespace simdjson {
+/**
+ * @brief cast a simdjson::ondemand::value to a std::array<int,4>
+ * useful for writing rx_roi from json header
+ */
+template <> simdjson_inline simdjson::simdjson_result<std::array<int, 4>> simdjson::ondemand::value::get() noexcept {
+    ondemand::array array;
+    auto error = get_array().get(array);
+    if (error) {
+        return error;
+    }
+    std::array<int, 4> arr;
+    int i = 0;
+    for (auto v : array) {
+        int64_t val;
+        error = v.get_int64().get(val);
+
+        if (error) {
+            return error;
+        }
+        arr[i++] = val;
+    }
+    return arr;
+}
+
+/**
+ * @brief cast a simdjson::ondemand::value to a uint32_t
+ * adds a check for 32bit overflow
+ */
+template <> simdjson_inline simdjson::simdjson_result<uint32_t> simdjson::ondemand::value::get() noexcept {
+    size_t val;
+    auto error = get_uint64().get(val);
+    if (error) {
+        return error;
+    }
+    if (val > std::numeric_limits<uint32_t>::max()) {
+        return 1;
+    }
+    return static_cast<uint32_t>(val);
+}
+
+} // namespace simdjson
+
+namespace aare {
+
+/** zmq header structure (from slsDetectorPackage)*/
+struct ZmqHeader {
+    /** true if incoming data, false if end of acquisition */
+    bool data{true};
+    uint32_t jsonversion{0};
+    uint32_t dynamicRange{0};
+    uint64_t fileIndex{0};
+    /** number of detectors/port in x axis */
+    uint32_t ndetx{0};
+    /** number of detectors/port in y axis */
+    uint32_t ndety{0};
+    /** number of pixels/channels in x axis for this zmq socket */
+    uint32_t npixelsx{0};
+    /** number of pixels/channels in y axis for this zmq socket */
+    uint32_t npixelsy{0};
+    /** number of bytes for an image in this socket */
+    uint32_t imageSize{0};
+    /** frame number from detector */
+    uint64_t acqIndex{0};
+    /** frame index (starting at 0 for each acquisition) */
+    uint64_t frameIndex{0};
+    /** progress in percentage */
+    double progress{0};
+    /** file name prefix */
+    std::string fname;
+    /** header from detector */
+    uint64_t frameNumber{0};
+    uint32_t expLength{0};
+    uint32_t packetNumber{0};
+    uint64_t detSpec1{0};
+    uint64_t timestamp{0};
+    uint16_t modId{0};
+    uint16_t row{0};
+    uint16_t column{0};
+    uint16_t detSpec2{0};
+    uint32_t detSpec3{0};
+    uint16_t detSpec4{0};
+    uint8_t detType{0};
+    uint8_t version{0};
+    /** if rows of image should be flipped */
+    int flipRows{0};
+    /** quad type (eiger hardware specific) */
+    uint32_t quad{0};
+    /** true if complete image, else missing packets */
+    bool completeImage{false};
+    /** additional json header */
+    std::map<std::string, std::string> addJsonHeader;
+    /** (xmin, xmax, ymin, ymax) roi only in files written */
+    std::array<int, 4> rx_roi{};
+
+    /** serialize struct to json string */
+    std::string to_string() const;
+    void from_string(std::string &s);
+};
+} // namespace aare

network_io/include/aare/ZmqSocket.hpp

@@ -0,0 +1,40 @@
+#pragma once
+
+#include <array>
+#include <map>
+#include <string>
+
+// Socket to receive data from a ZMQ publisher
+// needs to be in sync with the main library (or maybe better use the versioning in the header)
+
+// forward declare zmq_msg_t to avoid including zmq.h in the header
+class zmq_msg_t;
+
+namespace aare {
+
+class ZmqSocket {
+  protected:
+    void *m_context{nullptr};
+    void *m_socket{nullptr};
+    std::string m_endpoint;
+    int m_zmq_hwm{1000};
+    int m_timeout_ms{1000};
+    size_t m_potential_frame_size{1024 * 1024};
+    constexpr static size_t m_max_header_size = 1024;
+    char *m_header_buffer = new char[m_max_header_size];
+
+  public:
+    ZmqSocket() = default;
+    ~ZmqSocket();
+
+    ZmqSocket(const ZmqSocket &) = delete;
+    ZmqSocket operator=(const ZmqSocket &) = delete;
+    ZmqSocket(ZmqSocket &&) = delete;
+
+    void disconnect();
+    void set_zmq_hwm(int hwm);
+    void set_timeout_ms(int n);
+    void set_potential_frame_size(size_t size);
+};
+
+} // namespace aare

network_io/include/aare/ZmqSocketReceiver.hpp

@@ -1,5 +1,8 @@
 #pragma once
 
+#include "ZmqHeader.hpp"
+#include "ZmqSocket.hpp"
+
 #include <array>
 #include <cstdint>
 #include <map>
@@ -13,81 +16,11 @@ class zmq_msg_t;
 
 namespace aare {
 
-/** zmq header structure (from slsDetectorPackage)*/
-struct zmqHeader {
-    /** true if incoming data, false if end of acquisition */
-    bool data{true};
-    uint32_t jsonversion{0};
-    uint32_t dynamicRange{0};
-    uint64_t fileIndex{0};
-    /** number of detectors/port in x axis */
-    uint32_t ndetx{0};
-    /** number of detectors/port in y axis */
-    uint32_t ndety{0};
-    /** number of pixels/channels in x axis for this zmq socket */
-    uint32_t npixelsx{0};
-    /** number of pixels/channels in y axis for this zmq socket */
-    uint32_t npixelsy{0};
-    /** number of bytes for an image in this socket */
-    uint32_t imageSize{0};
-    /** frame number from detector */
-    uint64_t acqIndex{0};
-    /** frame index (starting at 0 for each acquisition) */
-    uint64_t frameIndex{0};
-    /** progress in percentage */
-    double progress{0};
-    /** file name prefix */
-    std::string fname;
-    /** header from detector */
-    uint64_t frameNumber{0};
-    uint32_t expLength{0};
-    uint32_t packetNumber{0};
-    uint64_t detSpec1{0};
-    uint64_t timestamp{0};
-    uint16_t modId{0};
-    uint16_t row{0};
-    uint16_t column{0};
-    uint16_t detSpec2{0};
-    uint32_t detSpec3{0};
-    uint16_t detSpec4{0};
-    uint8_t detType{0};
-    uint8_t version{0};
-    /** if rows of image should be flipped */
-    int flipRows{0};
-    /** quad type (eiger hardware specific) */
-    uint32_t quad{0};
-    /** true if complete image, else missing packets */
-    bool completeImage{false};
-    /** additional json header */
-    std::map<std::string, std::string> addJsonHeader;
-    /** (xmin, xmax, ymin, ymax) roi only in files written */
-    std::array<int, 4> rx_roi{};
-};
-
-class ZmqSocketReceiver {
-    void *m_context{nullptr};
-    void *m_socket{nullptr};
-    std::string m_endpoint;
-    int m_zmq_hwm{1000};
-    int m_timeout_ms{1000};
-    constexpr static size_t m_max_header_size = 1024;
-    char *m_header_buffer = new char[m_max_header_size];
-
-    bool decode_header(zmqHeader &h);
-
+class ZmqSocketReceiver : public ZmqSocket {
   public:
     ZmqSocketReceiver(const std::string &endpoint);
-    ~ZmqSocketReceiver();
-    ZmqSocketReceiver(const ZmqSocketReceiver &) = delete;
-    ZmqSocketReceiver operator=(const ZmqSocketReceiver &) = delete;
-    ZmqSocketReceiver(ZmqSocketReceiver &&) = delete;
-
     void connect();
-    void disconnect();
-    void set_zmq_hwm(int hwm);
-    void set_timeout_ms(int n);
-
-    int receive(zmqHeader &header, std::byte *data);
+    int receive(ZmqHeader &header, std::byte *data, bool serialized_header = false);
 };
 
 } // namespace aare

network_io/include/aare/ZmqSocketSender.hpp

@@ -0,0 +1,12 @@
+#pragma once
+#include "ZmqHeader.hpp"
+#include "ZmqSocket.hpp"
+
+namespace aare {
+class ZmqSocketSender : public ZmqSocket {
+  public:
+    ZmqSocketSender(const std::string &endpoint);
+    void bind();
+    int send(ZmqHeader &header, const std::byte *data, size_t size, bool serialize_header = false);
+};
+} // namespace aare