Add zmq buffer options to config file

3 files changed, 110 insertions, 55 deletions

diff --git a/doc/example.mux b/doc/example.mux
index 1dbec7d..26a231c 100644
--- a/doc/example.mux
+++ b/doc/example.mux
@@ -97,10 +97,6 @@ subchannels {
         type audio
         ; example file input
         inputfile "funk.mp2"
-        ; example zmq input:
-        ; accept connections to port 9001 from any interface
-        ; use toolame-dab as encoder
-        ;inputfile "tcp://*:9001"
         nonblock false
         bitrate 128
         id 10
@@ -119,24 +115,49 @@ subchannels {
         ; example file input
         ;inputfile "rick.dabp"
         ; example zmq input:
-        ; accept connections to port 9000 from any interface
-        ; use fdk-aac-dabplus-zmq as encoder
+        ; Accepts connections to port 9000 from any interface.
+        ; Use fdk-aac-dabplus as encoder
         inputfile "tcp://*:9000"
-        nonblock false
         bitrate 96
         id 1
         protection 1
+
+        ; ZMQ specific options, mandatory:
+
+        ; Maximum size of input buffer, in AAC frames (24ms)
+        ; when this buffer size is reached, some frames will be
+        ; discarded to get the size again below this value.
+        ; As the present implementation discards entire AAC superframes,
+        ; (5 frames = 120ms) the effect will clearly be audible.
+        zmq-buffer 40
+
+        ; At startup or after an underrun, the buffer is filled to this
+        ; amount of AAC frames before streaming starts.
+        zmq-prebuffering 20
+
+        ; In an ideal scenario, where the input rate exactly corresponds
+        ; to the rate at which the frames are consumed by dabmux, you
+        ; see the buffer level staying around the zmq-prebuffering value.
+        ; Network latency jitter can make it temporarily go lower or higher.
+        ; Encoder clock drift will make the buffer either slowly fill or
+        ; empty, which will create intermittent glitches.
     }
 
     sub-ri2 {
-        type dabplus
-        ; for dabplus types, you can use the ZeroMQ input (if compiled in)
-        ; with the following configuration:
-        inputfile "tcp://localhost:9000"
-        nonblock false
+        type audio
+        ; for audio types, you can use the ZeroMQ input (if compiled in)
+        ; with the following configuration in combination with
+        ; toolame-dab
+        ;
+        ; Support for toolame-dab is not as good as with fdk-aac-dabplus
+        inputfile "tcp://*:9001"
         bitrate 96
         id 1
         protection 1
+
+        ; The options are the same as for dabplus
+        zmq-buffer 40
+        zmq-prebuffering 20
     }
 }
 
diff --git a/src/ParserConfigfile.cpp b/src/ParserConfigfile.cpp
index f781b78..6b261e5 100644
--- a/src/ParserConfigfile.cpp
+++ b/src/ParserConfigfile.cpp
@@ -585,28 +585,43 @@ void setup_subchannel_from_ptree(dabSubchannel* subchan,
 #endif // defined(HAVE_INPUT_FILE)
 #if defined(HAVE_INPUT_ZEROMQ)
         }
-        else if (strcmp(subchan->inputProto, "tcp") == 0) {
+        else if ((strcmp(subchan->inputProto, "tcp") == 0) ||
+                (strcmp(subchan->inputProto, "epmg") == 0) ||
+                (strcmp(subchan->inputProto, "ipc") == 0) ) {
             input_is_old_style = false;
-            DabInputZmqMPEG* inzmq = new DabInputZmqMPEG(subchanuid);
-            inzmq->enrol_at(*rc);
-            subchan->input     = inzmq;
-            subchan->inputName = full_inputName;
-        }
-        else if (strcmp(subchan->inputProto, "epmg") == 0) {
-            etiLog.level(warn) << "Using untested epmg:// zeromq input";
-            input_is_old_style = false;
-            DabInputZmqMPEG* inzmq = new DabInputZmqMPEG(subchanuid);
-            inzmq->enrol_at(*rc);
-            subchan->input     = inzmq;
-            subchan->inputName = full_inputName;
-        }
-        else if (strcmp(subchan->inputProto, "ipc") == 0) {
-            etiLog.level(warn) << "Using untested ipc:// zeromq input";
-            input_is_old_style = false;
-            DabInputZmqMPEG* inzmq = new DabInputZmqMPEG(subchanuid);
+
+            int buffer_size;
+            int prebuffering;
+            try {
+                buffer_size = pt.get<int>("zmq-buffer");
+            }
+            catch (ptree_error &e) {
+                stringstream ss;
+                ss << "ZMQ Subchannel with uid " << subchanuid <<
+                    " has no zmq-buffer defined!";
+                throw runtime_error(ss.str());
+            }
+            try {
+                prebuffering = pt.get<int>("zmq-prebuffering");
+            }
+            catch (ptree_error &e) {
+                stringstream ss;
+                ss << "ZMQ Subchannel with uid " << subchanuid <<
+                    " has no zmq-buffer defined!";
+                throw runtime_error(ss.str());
+            }
+            DabInputZmqMPEG* inzmq =
+                new DabInputZmqMPEG(subchanuid, buffer_size, prebuffering);
             inzmq->enrol_at(*rc);
             subchan->input     = inzmq;
             subchan->inputName = full_inputName;
+
+            if (strcmp(subchan->inputProto, "epmg") == 0) {
+                etiLog.level(warn) << "Using untested epmg:// zeromq input";
+            }
+            else if (strcmp(subchan->inputProto, "ipc") == 0) {
+                etiLog.level(warn) << "Using untested ipc:// zeromq input";
+            }
 #endif // defined(HAVE_INPUT_ZEROMQ)
         } else {
             stringstream ss;
@@ -643,28 +658,43 @@ void setup_subchannel_from_ptree(dabSubchannel* subchan,
 #endif // defined(HAVE_INPUT_FILE)
 #if defined(HAVE_INPUT_ZEROMQ)
         }
-        else if (strcmp(subchan->inputProto, "tcp") == 0) {
+        else if ((strcmp(subchan->inputProto, "tcp") == 0) ||
+                (strcmp(subchan->inputProto, "epmg") == 0) ||
+                (strcmp(subchan->inputProto, "ipc") == 0) ) {
             input_is_old_style = false;
-            DabInputZmqAAC* inzmq = new DabInputZmqAAC(subchanuid);
-            inzmq->enrol_at(*rc);
-            subchan->input     = inzmq;
-            subchan->inputName = full_inputName;
-        }
-        else if (strcmp(subchan->inputProto, "epmg") == 0) {
-            etiLog.level(warn) << "Using untested epmg:// zeromq input";
-            input_is_old_style = false;
-            DabInputZmqAAC* inzmq = new DabInputZmqAAC(subchanuid);
-            inzmq->enrol_at(*rc);
-            subchan->input     = inzmq;
-            subchan->inputName = full_inputName;
-        }
-        else if (strcmp(subchan->inputProto, "ipc") == 0) {
-            etiLog.level(warn) << "Using untested ipc:// zeromq input";
-            input_is_old_style = false;
-            DabInputZmqAAC* inzmq = new DabInputZmqAAC(subchanuid);
+
+            int buffer_size;
+            int prebuffering;
+            try {
+                buffer_size = pt.get<int>("zmq-buffer");
+            }
+            catch (ptree_error &e) {
+                stringstream ss;
+                ss << "ZMQ Subchannel with uid " << subchanuid <<
+                    " has no zmq-buffer defined!";
+                throw runtime_error(ss.str());
+            }
+            try {
+                prebuffering = pt.get<int>("zmq-prebuffering");
+            }
+            catch (ptree_error &e) {
+                stringstream ss;
+                ss << "ZMQ Subchannel with uid " << subchanuid <<
+                    " has no zmq-buffer defined!";
+                throw runtime_error(ss.str());
+            }
+            DabInputZmqAAC* inzmq =
+                new DabInputZmqAAC(subchanuid, buffer_size, prebuffering);
             inzmq->enrol_at(*rc);
             subchan->input     = inzmq;
             subchan->inputName = full_inputName;
+
+            if (strcmp(subchan->inputProto, "epmg") == 0) {
+                etiLog.level(warn) << "Using untested epmg:// zeromq input";
+            }
+            else if (strcmp(subchan->inputProto, "ipc") == 0) {
+                etiLog.level(warn) << "Using untested ipc:// zeromq input";
+            }
 #endif // defined(HAVE_INPUT_ZEROMQ)
         } else {
             stringstream ss;
diff --git a/src/dabInputZmq.h b/src/dabInputZmq.h
index 11fb49a..a052815 100644
--- a/src/dabInputZmq.h
+++ b/src/dabInputZmq.h
@@ -76,13 +76,14 @@
 
 class DabInputZmqBase : public DabInputBase, public RemoteControllable {
     public:
-        DabInputZmqBase(const std::string name)
+        DabInputZmqBase(const std::string name,
+                int buffer_size, int prebuffering)
             : RemoteControllable(name),
             m_zmq_context(1),
             m_zmq_sock(m_zmq_context, ZMQ_SUB),
-            m_bitrate(0), m_prebuffering(INPUT_ZMQ_DEF_PREBUFFERING),
+            m_bitrate(0), m_prebuffering(prebuffering),
             m_enable_input(true),
-            m_frame_buffer_limit(INPUT_ZMQ_DEF_BUFFER_SIZE) {
+            m_frame_buffer_limit(buffer_size) {
                 RC_ADD_PARAMETER(enable,
                         "If the input is enabled. Set to zero to empty the buffer.");
             }
@@ -119,8 +120,9 @@ class DabInputZmqBase : public DabInputBase, public RemoteControllable {
 
 class DabInputZmqMPEG : public DabInputZmqBase {
     public:
-        DabInputZmqMPEG(const std::string name)
-            : DabInputZmqBase(name) {
+        DabInputZmqMPEG(const std::string name,
+                int buffer_size, int prebuffering)
+            : DabInputZmqBase(name, buffer_size, prebuffering) {
                 RC_ADD_PARAMETER(buffer,
                         "Size of the input buffer [mpeg frames]");
 
@@ -134,8 +136,9 @@ class DabInputZmqMPEG : public DabInputZmqBase {
 
 class DabInputZmqAAC : public DabInputZmqBase {
     public:
-        DabInputZmqAAC(const std::string name)
-            : DabInputZmqBase(name) {
+        DabInputZmqAAC(const std::string name,
+                int buffer_size, int prebuffering)
+            : DabInputZmqBase(name, buffer_size, prebuffering) {
                 RC_ADD_PARAMETER(buffer,
                         "Size of the input buffer [aac superframes]");
 
@@ -150,3 +153,4 @@ class DabInputZmqAAC : public DabInputZmqBase {
 #endif // HAVE_INPUT_ZMQ
 
 #endif // DAB_INPUT_ZMQ_H
+