add network_io folder

network_io/CMakeLists.txt

@@ -0,0 +1,15 @@
+add_library(network_io STATIC src/ZmqSocketReceiver.cpp)
+target_include_directories(network_io PUBLIC include)
+target_link_libraries(network_io PRIVATE libzmq fmt::fmt core utils aare_compiler_flags)
+
+if(AARE_PYTHON_BINDINGS)
+set_property(TARGET file_io PROPERTY POSITION_INDEPENDENT_CODE ON)
+endif()
+
+# if(AARE_TESTS)
+#     set(TestSources
+#             ${CMAKE_CURRENT_SOURCE_DIR}/test/NumpyFile.test.cpp
+#         )
+#     target_sources(tests PRIVATE ${TestSources} )
+#     target_link_libraries(tests PRIVATE core network_io)
+# endif()

network_io/include/aare/ZmqSocketReceiver.hpp

@@ -0,0 +1,93 @@
+#pragma once
+
+#include <array>
+#include <cstdint>
+#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 {
+
+/** 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);
+
+  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);
+};
+
+} // namespace aare

network_io/src/ZmqSocketReceiver.cpp

@@ -0,0 +1,82 @@
+#include "aare/ZmqSocketReceiver.hpp"
+#include <fmt/core.h>
+#include <zmq.h>
+
+namespace aare {
+
+ZmqSocketReceiver::ZmqSocketReceiver(const std::string &endpoint) : m_endpoint(endpoint) {
+    memset(m_header_buffer, 0, m_max_header_size);
+}
+
+void ZmqSocketReceiver::connect() {
+    m_context = zmq_ctx_new();
+    m_socket = zmq_socket(m_context, ZMQ_SUB);
+    fmt::print("Setting ZMQ_RCVHWM to {}\n", m_zmq_hwm);
+    int rc = zmq_setsockopt(m_socket, ZMQ_RCVHWM, &m_zmq_hwm, sizeof(m_zmq_hwm)); // should be set before connect
+    if (rc)
+        throw std::runtime_error(fmt::format("Could not set ZMQ_RCVHWM: {}", strerror(errno)));
+
+    int bufsize = 1024 * 1024 * m_zmq_hwm;
+    fmt::print("Setting ZMQ_RCVBUF to: {} MB\n", bufsize / (1024 * 1024));
+    rc = zmq_setsockopt(m_socket, ZMQ_RCVBUF, &bufsize, sizeof(bufsize));
+    if (rc)
+        throw std::runtime_error(fmt::format("Could not set ZMQ_RCVBUF: {}", strerror(errno)));
+
+    zmq_connect(m_socket, m_endpoint.c_str());
+    zmq_setsockopt(m_socket, ZMQ_SUBSCRIBE, "", 0);
+}
+
+void ZmqSocketReceiver::disconnect() {
+    zmq_close(m_socket);
+    zmq_ctx_destroy(m_context);
+    m_socket = nullptr;
+    m_context = nullptr;
+}
+
+ZmqSocketReceiver::~ZmqSocketReceiver() {
+    if (m_socket)
+        disconnect();
+    delete[] m_header_buffer;
+}
+
+void ZmqSocketReceiver::set_zmq_hwm(int hwm) { m_zmq_hwm = hwm; }
+
+void ZmqSocketReceiver::set_timeout_ms(int n) { m_timeout_ms = n; }
+
+int ZmqSocketReceiver::receive(zmqHeader &header, std::byte *data) {
+
+    // receive header
+    int header_bytes_received = zmq_recv(m_socket, m_header_buffer, m_max_header_size, 0);
+    m_header_buffer[header_bytes_received] = '\0'; // make sure we zero terminate
+    if (header_bytes_received < 0) {
+        fmt::print("Error receiving header: {}\n", strerror(errno));
+        return -1;
+    }
+    fmt::print("Bytes: {}, Header: {}\n", header_bytes_received, m_header_buffer);
+
+    // decode header
+    if (!decode_header(header)) {
+        fmt::print("Error decoding header\n");
+        return -1;
+    }
+
+    // do we have a multipart message (data following header)?
+    int more;
+    size_t more_size = sizeof(more);
+    zmq_getsockopt(m_socket, ZMQ_RCVMORE, &more, &more_size);
+    if (!more) {
+        return 0; // no data following header
+    } else {
+        int data_bytes_received = zmq_recv(m_socket, data, 1024 * 1024 * 2, 0); // TODO! configurable size!!!!
+        if (data_bytes_received == -1)
+            throw std::runtime_error("Got half of a multipart msg!!!");
+    }
+    return 1;
+}
+
+bool ZmqSocketReceiver::decode_header(zmqHeader &h) {
+    // TODO: implement
+    return true;
+}
+
+} // namespace aare