From fa36e03255eb668195f043d39307f3dc2fa5e809 Mon Sep 17 00:00:00 2001
From: "Matthias P. Braendli" <matthias.braendli@mpb.li>
Date: Fri, 11 Apr 2014 11:55:43 +0200
Subject: Add zmq buffer options to config file

---
 src/ParserConfigfile.cpp | 102 ++++++++++++++++++++++++++++++-----------------
 1 file changed, 66 insertions(+), 36 deletions(-)

(limited to 'src/ParserConfigfile.cpp')

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;
-- 
cgit v1.2.3