Add ZMQ output to odr-zmq2edi

3 files changed, 125 insertions, 53 deletions

diff --git a/src/zmq2edi/EDISender.cpp b/src/zmq2edi/Sender.cpp
index 4a70105..fe46846 100644
--- a/src/zmq2edi/EDISender.cpp
+++ b/src/zmq2edi/Sender.cpp
@@ -3,7 +3,7 @@
    2011, 2012 Her Majesty the Queen in Right of Canada (Communications
    Research Center Canada)
 
-   Copyright (C) 2020
+   Copyright (C) 2024
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -25,7 +25,7 @@
    along with ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-#include "EDISender.h"
+#include "Sender.h"
 #include "Log.h"
 #include <cmath>
 #include <numeric>
@@ -35,20 +35,27 @@
 
 using namespace std;
 
-EDISender::~EDISender()
+Sender::Sender() :
+    zmq_ctx(2)
+{
+}
+
+Sender::~Sender()
 {
     if (running.load()) {
         running.store(false);
 
         // Unblock thread
         frame_t emptyframe;
-        frames.push(emptyframe);
+        frames.push(std::move(emptyframe));
 
         process_thread.join();
     }
 }
 
-void EDISender::start(const edi::configuration_t& conf, int delay_ms, bool drop_late_packets)
+void Sender::start(const edi::configuration_t& conf,
+        const zmq_send_config_t& zmq_conf,
+        int delay_ms, bool drop_late_packets)
 {
     edi_conf = conf;
     tist_delay_ms = delay_ms;
@@ -56,16 +63,22 @@ void EDISender::start(const edi::configuration_t& conf, int delay_ms, bool drop_
 
     edi_sender = make_shared<edi::Sender>(edi_conf);
 
+    for (const auto& url : zmq_conf.urls) {
+        zmq::socket_t zmq_sock(zmq_ctx, ZMQ_PUB);
+        zmq_sock.bind(url.c_str());
+        zmq_sockets.emplace_back(std::move(zmq_sock));
+    }
+
     running.store(true);
-    process_thread = thread(&EDISender::process, this);
+    process_thread = thread(&Sender::process, this);
 }
 
-void EDISender::push_frame(frame_t&& frame)
+void Sender::push_frame(frame_t&& frame)
 {
-    frames.push(move(frame));
+    frames.push(std::move(frame));
 }
 
-void EDISender::print_configuration()
+void Sender::print_configuration()
 {
     if (edi_conf.enabled()) {
         edi_conf.print();
@@ -75,7 +88,7 @@ void EDISender::print_configuration()
     }
 }
 
-void EDISender::send_eti_frame(frame_t& frame)
+void Sender::send_eti_frame(frame_t& frame)
 {
     uint8_t *p = frame.data.data();
 
@@ -221,9 +234,18 @@ void EDISender::send_eti_frame(frame_t& frame)
 
         edi_sender->write(edi_tagpacket);
     }
+
+    if (not frame.original_zmq_message.empty()) {
+        for (auto& sock : zmq_sockets) {
+            const auto send_result = sock.send(frame.original_zmq_message, zmq::send_flags::dontwait);
+            if (not send_result.has_value()) {
+                num_zmq_send_errors++;
+            }
+        }
+    }
 }
 
-void EDISender::process()
+void Sender::process()
 {
     while (running.load()) {
         frame_t frame;
@@ -289,7 +311,8 @@ void EDISender::process()
                 " stdev: " << stdev <<
                 " late: " <<
                 num_late << " of " << buffering_stats.size() << " (" <<
-                num_late * 100.0 / n << "%)";
+                num_late * 100.0 / n << "%) " <<
+                "Num ZMQ send errors: " << num_zmq_send_errors;
 
             buffering_stats.clear();
         }
diff --git a/src/zmq2edi/EDISender.h b/src/zmq2edi/Sender.h
index 9c37e75..6dfd615 100644
--- a/src/zmq2edi/EDISender.h
+++ b/src/zmq2edi/Sender.h
@@ -3,7 +3,7 @@
    2011, 2012 Her Majesty the Queen in Right of Canada (Communications
    Research Center Canada)
 
-   Copyright (C) 2020
+   Copyright (C) 2024
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -46,18 +46,27 @@ struct metadata_t {
 };
 
 struct frame_t {
+    // Since a zmq message actually contains 4 frames, the
+    // original_zmq_msg is only non-empty for the first of the
+    // four calls to Sender::send_edi_frame().
+    zmq::message_t original_zmq_message;
     std::vector<uint8_t> data;
     metadata_t metadata;
     std::chrono::steady_clock::time_point received_at;
 };
 
-class EDISender {
+struct zmq_send_config_t {
+    std::vector<std::string> urls;
+};
+
+class Sender {
     public:
-        EDISender() = default;
-        EDISender(const EDISender& other) = delete;
-        EDISender& operator=(const EDISender& other) = delete;
-        ~EDISender();
+        Sender();
+        Sender(const Sender& other) = delete;
+        Sender& operator=(const Sender& other) = delete;
+        ~Sender();
         void start(const edi::configuration_t& conf,
+                const zmq_send_config_t& zmq_conf,
                 int delay_ms, bool drop_late_packets);
         void push_frame(frame_t&& frame);
         void print_configuration(void);
@@ -75,11 +84,15 @@ class EDISender {
 
         std::shared_ptr<edi::Sender> edi_sender;
 
+        zmq::context_t zmq_ctx;
+        std::vector<zmq::socket_t> zmq_sockets;
+
         struct buffering_stat_t {
             // Time between when we received the packets and when we transmit packets, in microseconds
             double buffering_time_us = 0.0;
             bool late = false;
         };
         std::vector<buffering_stat_t> buffering_stats;
+        size_t num_zmq_send_errors = 0;
 
 };
diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp
index cff16c7..41d92b5 100644
--- a/src/zmq2edi/zmq2edi.cpp
+++ b/src/zmq2edi/zmq2edi.cpp
@@ -3,7 +3,7 @@
    2011, 2012 Her Majesty the Queen in Right of Canada (Communications
    Research Center Canada)
 
-   Copyright (C) 2020
+   Copyright (C) 2024
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -36,7 +36,7 @@
 #include <thread>
 #include <vector>
 
-#include "EDISender.h"
+#include "Sender.h"
 #include "dabOutput/dabOutput.h"
 
 constexpr size_t MAX_ERROR_COUNT = 10;
@@ -45,7 +45,7 @@ constexpr long DEFAULT_BACKOFF = 5000;
 
 static edi::configuration_t edi_conf;
 
-static EDISender edisender;
+static Sender edisender;
 
 static void usage()
 {
@@ -54,6 +54,8 @@ static void usage()
     cerr << "Usage:" << endl;
     cerr << "odr-zmq2edi [options] <source>" << endl << endl;
 
+    cerr << "ODR-ZMQ2EDI can output to both EDI and ZMQ. It buffers and releases frames according to their timestamp." << endl;
+
     cerr << "Options:" << endl;
     cerr << "The following options can be given only once:" << endl;
     cerr << " <source> is a ZMQ URL that points to a ODR-DabMux ZMQ output." << endl;
@@ -62,15 +64,20 @@ static void usage()
     cerr << " -C <path to script>   Before starting, run the given script, and only start if it returns 0." << endl;
     cerr << "                       This is useful for checking that NTP is properly synchronised" << endl;
     cerr << " -x                    Drop frames where for which the wait time would be negative, i.e. frames that arrived too late." << endl;
+    cerr << " -b <backoff>          Number of milliseconds to backoff after an input reset (default " << DEFAULT_BACKOFF << ")." << endl << endl;
+
+    cerr << " ZMQ Output options:" << endl;
+    cerr << " -Z <url>              Add a zmq output to URL, e.g. tcp:*:9876 to listen for connections on port 9876 " << endl << endl;
+
+    cerr << " EDI Output options:" << endl;
+    cerr << " -v                    Enables verbose mode." << endl;
     cerr << " -P                    Disable PFT and send AFPackets." << endl;
     cerr << " -f <fec>              Set the FEC." << endl;
     cerr << " -i <spread>           Configure the UDP packet spread/interleaver with given percentage: 0% send all fragments at once, 100% spread over 24ms, >100% spread and interleave. Default 95%\n";
     cerr << " -D                    Dump the EDI to edi.debug file." << endl;
-    cerr << " -v                    Enables verbose mode." << endl;
     cerr << " -a <alignement>       Set the alignment of the TAG Packet (default 8)." << endl;
-    cerr << " -b <backoff>          Number of milliseconds to backoff after an input reset (default " << DEFAULT_BACKOFF << ")." << endl << endl;
 
-    cerr << "The following options can be given several times, when more than UDP destination is desired:" << endl;
+    cerr << "The following options can be given several times, when more than EDI/UDP destination is desired:" << endl;
     cerr << " -d <destination ip>   Set the destination ip." << endl;
     cerr << " -p <destination port> Set the destination port." << endl;
     cerr << " -s <source port>      Set the source port." << endl;
@@ -176,7 +183,7 @@ static void add_edi_destination(void)
                 std::to_string(edi_conf.destinations.size() + 1));
     }
 
-    edi_conf.destinations.push_back(move(edi_destination));
+    edi_conf.destinations.push_back(std::move(edi_destination));
     edi_destination = std::make_shared<edi::udp_destination_t>();
 
     dest_port_set = false;
@@ -249,9 +256,11 @@ int start(int argc, char **argv)
     uint32_t backoff_after_reset_ms = DEFAULT_BACKOFF;
     std::string startupcheck;
 
+    zmq_send_config_t zmq_conf;
+
     int ch = 0;
     while (ch != -1) {
-        ch = getopt(argc, argv, "C:d:p:s:S:t:Pf:i:Dva:b:w:xh");
+        ch = getopt(argc, argv, "C:d:p:s:S:t:Pf:i:Dva:b:w:xhZ:");
         switch (ch) {
             case -1:
                 break;
@@ -302,6 +311,9 @@ int start(int argc, char **argv)
             case 'x':
                 drop_late_packets = true;
                 break;
+            case 'Z':
+                zmq_conf.urls.push_back(optarg);
+                break;
             case 'h':
             default:
                 usage();
@@ -309,6 +321,32 @@ int start(int argc, char **argv)
         }
     }
 
+    if (dest_addr_set) {
+        add_edi_destination();
+    }
+
+    if (optind >= argc) {
+        etiLog.level(error) << "source option is missing";
+        return 1;
+    }
+
+    if (edi_conf.destinations.empty() and zmq_conf.urls.empty()) {
+        etiLog.level(error) << "No destinations set";
+        return 1;
+    }
+
+    if (not edi_conf.destinations.empty()) {
+        edisender.print_configuration();
+    }
+
+    if (not zmq_conf.urls.empty()) {
+        etiLog.level(info) << "Setting up ZMQ to:";
+        for (const auto& url : zmq_conf.urls) {
+            etiLog.level(info) << " " << url;
+        }
+    }
+
+
     if (not startupcheck.empty()) {
         etiLog.level(info) << "Running startup check '" << startupcheck << "'";
         int wstatus = system(startupcheck.c_str());
@@ -328,22 +366,9 @@ int start(int argc, char **argv)
         }
     }
 
-    add_edi_destination();
-
-    if (optind >= argc) {
-        etiLog.level(error) << "source option is missing";
-        return 1;
-    }
-
-    if (edi_conf.destinations.empty()) {
-        etiLog.level(error) << "No EDI destinations set";
-        return 1;
-    }
-
-    etiLog.level(info) << "Setting up EDI Sender with delay " << delay_ms << " ms. " <<
+    etiLog.level(info) << "Setting up Sender with delay " << delay_ms << " ms. " <<
         (drop_late_packets ? "Will" : "Will not") << " drop late packets";
-    edisender.start(edi_conf, delay_ms, drop_late_packets);
-    edisender.print_configuration();
+    edisender.start(edi_conf, zmq_conf, delay_ms, drop_late_packets);
 
     const char* source_url = argv[optind];
 
@@ -370,26 +395,33 @@ int start(int argc, char **argv)
                 }
                 else {
                     // Event received: recv will not block
-                    zmq_sock.recv(incoming, zmq::recv_flags::none);
+                    const auto recv_result = zmq_sock.recv(incoming, zmq::recv_flags::none);
+                    if (not recv_result.has_value()) {
+                        continue;
+                    }
+
                     const auto received_at = std::chrono::steady_clock::now();
 
-                    zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
+                    // Casting incoming.data() to zmq_dab_message_t* is not allowed, because
+                    // it might be misaligned
+                    zmq_dab_message_t dab_msg;
+                    memcpy(&dab_msg, incoming.data(), ZMQ_DAB_MESSAGE_HEAD_LENGTH);
 
-                    if (dab_msg->version != 1) {
-                        etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
+                    if (dab_msg.version != 1) {
+                        etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg.version;
                         error_count++;
                     }
 
-                    int offset = sizeof(dab_msg->version) +
-                        NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
+                    int offset = sizeof(dab_msg.version) +
+                        NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg.buflen);
 
                     std::vector<frame_t> all_frames;
                     all_frames.reserve(NUM_FRAMES_PER_ZMQ_MESSAGE);
 
                     for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
-                        if (dab_msg->buflen[i] <= 0 or dab_msg->buflen[i] > 6144) {
+                        if (dab_msg.buflen[i] <= 0 or dab_msg.buflen[i] > 6144) {
                             etiLog.level(error) << "ZeroMQ buffer " << i <<
-                                " has invalid length " << dab_msg->buflen[i];
+                                " has invalid length " << dab_msg.buflen[i];
                             error_count++;
                         }
                         else {
@@ -397,7 +429,7 @@ int start(int argc, char **argv)
                             frame.data.resize(6144, 0x55);
                             frame.received_at = received_at;
 
-                            const int framesize = dab_msg->buflen[i];
+                            const int framesize = dab_msg.buflen[i];
 
                             memcpy(frame.data.data(),
                                     ((uint8_t*)incoming.data()) + offset,
@@ -429,6 +461,10 @@ int start(int argc, char **argv)
                         offset += consumed_bytes;
                     }
 
+                    if (not all_frames.empty()) {
+                        all_frames[0].original_zmq_message = std::move(incoming);
+                    }
+
                     for (auto &f : all_frames) {
                         edisender.push_frame(std::move(f));
                     }
@@ -476,9 +512,6 @@ int main(int argc, char **argv)
 
     try {
         ret = start(argc, argv);
-
-        // To make sure things get printed to stderr
-        std::this_thread::sleep_for(std::chrono::milliseconds(300));
     }
     catch (const std::runtime_error &e) {
         etiLog.level(error) << "Runtime error: " << e.what();
@@ -487,5 +520,8 @@ int main(int argc, char **argv)
         etiLog.level(error) << "Logic error! " << e.what();
     }
 
+    // To make sure things get printed to stderr
+    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+
     return ret;
 }