Merge branch 'outputRefactoring' into next

1 files changed, 65 insertions, 49 deletions

diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp
index 783f0f5..f6a816a 100644
--- a/src/InputZeroMQReader.cpp
+++ b/src/InputZeroMQReader.cpp
@@ -3,7 +3,7 @@
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
 
-   Copyright (C) 2017
+   Copyright (C) 2018
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -68,29 +68,35 @@ struct zmq_dab_message_t
 #define ZMQ_DAB_MESSAGE_T_HEADERSIZE \
     (sizeof(uint32_t) + NUM_FRAMES_PER_ZMQ_MESSAGE*sizeof(uint16_t))
 
+InputZeroMQReader::~InputZeroMQReader()
+{
+    m_running = false;
+    m_zmqcontext.close();
+    if (m_recv_thread.joinable()) {
+            m_recv_thread.join();
+    }
+}
+
 int InputZeroMQReader::Open(const string& uri, size_t max_queued_frames)
 {
     // The URL might start with zmq+tcp://
     if (uri.substr(0, 4) == "zmq+") {
-        uri_ = uri.substr(4);
+        m_uri = uri.substr(4);
     }
     else {
-        uri_ = uri;
+        m_uri = uri;
     }
 
-    workerdata_.uri = uri_;
-    workerdata_.max_queued_frames = max_queued_frames;
-    // launch receiver thread
-    worker_.Start(&workerdata_);
+    m_max_queued_frames = max_queued_frames;
+
+    m_recv_thread = boost::thread(&InputZeroMQReader::RecvProcess, this);
 
     return 0;
 }
 
 int InputZeroMQReader::GetNextFrame(void* buffer)
 {
-    const size_t framesize = 6144;
-
-    if (not worker_.is_running()) {
+    if (not m_running) {
         return 0;
     }
 
@@ -100,77 +106,100 @@ int InputZeroMQReader::GetNextFrame(void* buffer)
      * (4 ETI frames in TM1) and we should make sure that
      * we can serve the data required for a full transmission frame.
      */
-    if (in_messages_.size() < 4) {
+    if (m_in_messages.size() < 4) {
         const size_t prebuffering = 10;
         etiLog.log(trace, "ZMQ,wait1");
-        in_messages_.wait_and_pop(incoming, prebuffering);
+        m_in_messages.wait_and_pop(incoming, prebuffering);
     }
     else {
         etiLog.log(trace, "ZMQ,wait2");
-        in_messages_.wait_and_pop(incoming);
+        m_in_messages.wait_and_pop(incoming);
     }
     etiLog.log(trace, "ZMQ,pop");
 
-    if (not worker_.is_running()) {
+    if (not m_running) {
         throw zmq_input_overflow();
     }
 
-    memcpy(buffer, &incoming->front(), framesize);
+
+    const size_t framesize = 6144;
+    if (incoming->empty()) {
+        return 0;
+    }
+    else if (incoming->size() == framesize) {
+        memcpy(buffer, &incoming->front(), framesize);
+    }
+    else {
+        throw logic_error("ZMQ ETI not 6144");
+    }
 
     return framesize;
 }
 
-void InputZeroMQReader::PrintInfo()
+void InputZeroMQReader::PrintInfo() const
 {
     fprintf(stderr, "Input ZeroMQ:\n");
-    fprintf(stderr, "  Receiving from %s\n\n", uri_.c_str());
+    fprintf(stderr, "  Receiving from %s\n\n", m_uri.c_str());
 }
 
-// ------------- Worker functions
-
-void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
+void InputZeroMQReader::RecvProcess()
 {
     set_thread_name("zmqinput");
-    size_t queue_size = 0;
+    m_running = true;
 
+    size_t queue_size = 0;
     bool buffer_full = false;
 
-    zmq::socket_t subscriber(zmqcontext, ZMQ_SUB);
+    zmq::socket_t subscriber(m_zmqcontext, ZMQ_SUB);
     // zmq sockets are not thread safe. That's why
     // we create it here, and not at object creation.
 
     bool success = true;
 
     try {
-        subscriber.connect(workerdata->uri.c_str());
+        subscriber.connect(m_uri.c_str());
     }
     catch (zmq::error_t& err) {
-        etiLog.level(error) << "Failed to connect ZeroMQ socket to '" << workerdata->uri << "': '" << err.what() << "'";
+        etiLog.level(error) << "Failed to connect ZeroMQ socket to '" <<
+            m_uri << "': '" << err.what() << "'";
         success = false;
     }
 
     if (success) try {
-        subscriber.setsockopt(ZMQ_SUBSCRIBE, NULL, 0); // subscribe to all messages
+        // subscribe to all messages
+        subscriber.setsockopt(ZMQ_SUBSCRIBE, NULL, 0);
     }
     catch (zmq::error_t& err) {
-        etiLog.level(error) << "Failed to subscribe ZeroMQ socket to messages: '" << err.what() << "'";
+        etiLog.level(error) << "Failed to subscribe ZeroMQ socket to messages: '" <<
+            err.what() << "'";
         success = false;
     }
 
     if (success) try {
-        while (running)
-        {
+        while (m_running) {
             zmq::message_t incoming;
+            zmq::pollitem_t items[1];
+            items[0].socket = subscriber;
+            items[0].events = ZMQ_POLLIN;
+            const int zmq_timeout_ms = 100;
+            const int num_events = zmq::poll(items, 1, zmq_timeout_ms);
+            if (num_events == 0) {
+                // timeout is signalled by an empty buffer
+                auto buf = make_shared<vector<uint8_t> >();
+                m_in_messages.push(buf);
+                continue;
+            }
+
             subscriber.recv(&incoming);
 
             if (m_to_drop) {
-                queue_size = workerdata->in_messages->size();
+                queue_size = m_in_messages.size();
                 if (queue_size > 4) {
-                    workerdata->in_messages->notify();
+                    m_in_messages.notify();
                 }
                 m_to_drop--;
             }
-            else if (queue_size < workerdata->max_queued_frames) {
+            else if (queue_size < m_max_queued_frames) {
                 if (buffer_full) {
                     etiLog.level(info) << "ZeroMQ buffer recovered: " <<
                         queue_size << " elements";
@@ -214,14 +243,14 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
 
                             offset += framesize;
 
-                            queue_size = workerdata->in_messages->push(buf);
+                            queue_size = m_in_messages.push(buf);
                             etiLog.log(trace, "ZMQ,push %zu", queue_size);
                         }
                     }
                 }
             }
             else {
-                workerdata->in_messages->notify();
+                m_in_messages.notify();
 
                 if (!buffer_full) {
                     etiLog.level(warn) << "ZeroMQ buffer overfull !";
@@ -230,7 +259,7 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
                     throw runtime_error("ZMQ input full");
                 }
 
-                queue_size = workerdata->in_messages->size();
+                queue_size = m_in_messages.size();
 
                 /* Drop three more incoming ETI frames before
                  * we start accepting them again, to guarantee
@@ -256,21 +285,8 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
 
     subscriber.close();
 
-    running = false;
-    workerdata->in_messages->notify();
-}
-
-void InputZeroMQWorker::Start(struct InputZeroMQThreadData* workerdata)
-{
-    running = true;
-    recv_thread = boost::thread(&InputZeroMQWorker::RecvProcess, this, workerdata);
-}
-
-void InputZeroMQWorker::Stop()
-{
-    running = false;
-    zmqcontext.close();
-    recv_thread.join();
+    m_running = false;
+    m_in_messages.notify();
 }
 
 #endif