Add new zeromq config format

This gives the choice for the transmission of the metadata,
and the enabling of the TAI downloader is also triggered
appropriately

7 files changed, 86 insertions, 27 deletions

diff --git a/doc/advanced.mux b/doc/advanced.mux
index 801857b..c3e216a 100644
--- a/doc/advanced.mux
+++ b/doc/advanced.mux
@@ -342,8 +342,8 @@ outputs {
     ; Please see doc/dab_output_formats.txt
     stdout "fifo:///dev/stdout?type=raw"
 
-    ; ZeroMQ output example
-    ; Listen on all interfaces, on port 9100
+    ; Legacy format for ZeroMQ output example. See example.mux
+    ; for newer format.
     ;zmq  "zmq+tcp://*:9100"
 
     ; Throttle output to real-time (one ETI frame every 24ms)
diff --git a/doc/example.mux b/doc/example.mux
index d764be3..ad2f3c8 100644
--- a/doc/example.mux
+++ b/doc/example.mux
@@ -167,10 +167,20 @@ outputs {
     ; Output RAW ETI NI to standard output
     stdout "fifo:///dev/stdout?type=raw"
 
-    ; ZeroMQ output example 
+    ; ZeroMQ output example, new configuration format. Several
+    ; zeromq blocks can be added here.
     ; This output does not back-pressure the multiplexer.
-    ; Listen on all interfaces, on port 9100
-    ;zmq  "zmq+tcp://*:9100"
+    zeromq {
+        ; Listen on all interfaces, on port 9100
+        endpoint "tcp://*:9100"
+
+        ; Transmit backward compatible metadata containing
+        ; EDI time and UTC offset when TIST is enabled.
+        ; WARNING! requires ODR-DabMux to be compiled with
+        ; EDI output, and this will enable leap second download
+        ; as for the EDI output!
+        allowmetadata true
+    }
 
     ; Output ETI-over-TCP. This is like piping a RAW ETI NI data stream
     ; into a TCP socket, except that the output can handle simultaneous
diff --git a/src/DabMultiplexer.cpp b/src/DabMultiplexer.cpp
index 58a1c8f..c109919 100644
--- a/src/DabMultiplexer.cpp
+++ b/src/DabMultiplexer.cpp
@@ -3,7 +3,7 @@
    2011, 2012 Her Majesty the Queen in Right of Canada (Communications
    Research Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
    */
 /*
@@ -75,6 +75,7 @@ DabMultiplexer::DabMultiplexer(
     sync(0x49C5F8),
     currentFrame(0),
     ensemble(std::make_shared<dabEnsemble>()),
+    m_tai_clock_required(false),
     m_clock_tai(),
     fig_carousel(ensemble)
 {
@@ -128,7 +129,7 @@ void DabMultiplexer::set_edi_config(const edi_configuration_t& new_edi_conf)
 
 
 // Run a set of checks on the configuration
-void DabMultiplexer::prepare()
+void DabMultiplexer::prepare(bool require_tai_clock)
 {
     parse_ptree(m_pt, ensemble);
 
@@ -168,17 +169,16 @@ void DabMultiplexer::prepare()
 
     bool tist_enabled = m_pt.get("general.tist", false);
 
-    if (tist_enabled and edi_conf.enabled()) {
+    m_tai_clock_required = (tist_enabled and edi_conf.enabled()) or require_tai_clock;
+
+    if (m_tai_clock_required) {
         try {
             m_clock_tai.get_offset();
         }
         catch (std::runtime_error& e) {
-            const char* err_msg =
-                "Could not initialise TAI clock properly required by "
-                "EDI with timestamp. Do you have a working internet "
-                "connection?";
-
-            etiLog.level(error) << err_msg;
+            etiLog.level(error) <<
+                "Could not initialise TAI clock properly. "
+                "Do you have a working internet connection?";
             throw;
         }
     }
@@ -654,16 +654,16 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
     try {
         bool tist_enabled = m_pt.get("general.tist", false);
 
-        if (tist_enabled and edi_conf.enabled()) {
+        if (tist_enabled and m_tai_clock_required) {
             edi_tagDETI.set_seconds(edi_time);
 
             // In case get_offset fails, we still want to update the EDI seconds
-            const auto utco = m_clock_tai.get_offset();
-            edi_tagDETI.set_tai_utc_offset(utco);
+            const auto tai_utc_offset = m_clock_tai.get_offset();
+            edi_tagDETI.set_tai_utc_offset(tai_utc_offset);
 
             for (auto output : outputs) {
                 shared_ptr<OutputMetadata> md_utco =
-                    make_shared<OutputMetadataUTCO>(utco);
+                    make_shared<OutputMetadataUTCO>(edi_tagDETI.utco);
                 output->setMetadata(md_utco);
 
                 shared_ptr<OutputMetadata> md_edi_time =
diff --git a/src/DabMultiplexer.h b/src/DabMultiplexer.h
index b3e432e..80b3ab9 100644
--- a/src/DabMultiplexer.h
+++ b/src/DabMultiplexer.h
@@ -56,7 +56,7 @@ class DabMultiplexer : public RemoteControllable {
     public:
         DabMultiplexer(
                 boost::property_tree::ptree pt);
-        void prepare(void);
+        void prepare(bool require_tai_clock);
 
         unsigned long getCurrentFrame() { return currentFrame; }
 
@@ -92,6 +92,7 @@ class DabMultiplexer : public RemoteControllable {
 
         std::shared_ptr<dabEnsemble> ensemble;
 
+        bool m_tai_clock_required;
         ClockTAI m_clock_tai;
 
 #if HAVE_OUTPUT_EDI
diff --git a/src/DabMux.cpp b/src/DabMux.cpp
index 5d109fa..c0f5ad0 100644
--- a/src/DabMux.cpp
+++ b/src/DabMux.cpp
@@ -275,6 +275,7 @@ int main(int argc, char *argv[])
 
         /******************** READ OUTPUT PARAMETERS ***************/
         set<string> all_output_names;
+        bool output_require_tai_clock = false;
         ptree pt_outputs = pt.get_child("outputs");
         for (auto ptree_pair : pt_outputs) {
             string outputuid = ptree_pair.first;
@@ -333,6 +334,46 @@ int main(int argc, char *argv[])
                 throw runtime_error("EDI output not compiled in");
 #endif
             }
+            else if (outputuid == "zeromq") {
+#if defined(HAVE_OUTPUT_ZEROMQ)
+                ptree pt_zeromq = pt_outputs.get_child("zeromq");
+                shared_ptr<DabOutput> output;
+
+                string endpoint = pt_zeromq.get<string>("endpoint");
+                bool allow_metadata = pt_zeromq.get<bool>("allowmetadata");
+                output_require_tai_clock |= allow_metadata;
+
+                size_t proto_pos = endpoint.find("://");
+                if (proto_pos == std::string::npos) {
+                    stringstream ss;
+                    ss << "zeromq output endpoint '" << endpoint << "' has incorrect format!";
+                    throw runtime_error(ss.str());
+                }
+
+                string proto = endpoint.substr(0, proto_pos);
+                string location = endpoint.substr(proto_pos + 3);
+
+                output = make_shared<DabOutputZMQ>(proto, allow_metadata);
+
+                if (not output) {
+                    etiLog.level(error) <<
+                        "Unable to init zeromq output " <<
+                        endpoint;
+                    return -1;
+                }
+
+                if (output->Open(location) == -1) {
+                    etiLog.level(error) <<
+                        "Unable to open zeromq output " <<
+                        endpoint;
+                    return -1;
+                }
+
+                outputs.push_back(output);
+#else
+                throw runtime_error("ZeroMQ output not compiled in");
+#endif
+            }
             else {
                 string uri = pt_outputs.get<string>(outputuid);
 
@@ -374,14 +415,17 @@ int main(int argc, char *argv[])
                     output = make_shared<DabOutputSimul>();
 #endif // defined(HAVE_OUTPUT_SIMUL)
 #if defined(HAVE_OUTPUT_ZEROMQ)
+                /* The legacy configuration setting will not enable metadata,
+                 * to keep backward compatibility
+                 */
                 } else if (proto == "zmq+tcp") {
-                    output = make_shared<DabOutputZMQ>("tcp");
+                    output = make_shared<DabOutputZMQ>("tcp", false);
                 } else if (proto == "zmq+ipc") {
-                    output = make_shared<DabOutputZMQ>("ipc");
+                    output = make_shared<DabOutputZMQ>("ipc", false);
                 } else if (proto == "zmq+pgm") {
-                    output = make_shared<DabOutputZMQ>("pgm");
+                    output = make_shared<DabOutputZMQ>("pgm", false);
                 } else if (proto == "zmq+epgm") {
-                    output = make_shared<DabOutputZMQ>("epgm");
+                    output = make_shared<DabOutputZMQ>("epgm", false);
 #endif // defined(HAVE_OUTPUT_ZEROMQ)
                 } else {
                     etiLog.level(error) << "Output protocol unknown: " << proto;
@@ -413,7 +457,7 @@ int main(int argc, char *argv[])
             throw MuxInitException();
         }
 
-        mux.prepare();
+        mux.prepare(output_require_tai_clock);
         mux.print_info();
 
         etiLog.log(info, "--- Output list ---");
diff --git a/src/dabOutput/dabOutput.h b/src/dabOutput/dabOutput.h
index 58a2929..4142b25 100644
--- a/src/dabOutput/dabOutput.h
+++ b/src/dabOutput/dabOutput.h
@@ -305,11 +305,12 @@ struct zmq_dab_message_t
 class DabOutputZMQ : public DabOutput
 {
     public:
-        DabOutputZMQ(const std::string &zmq_proto) :
+        DabOutputZMQ(const std::string &zmq_proto, bool allow_metadata) :
             endpoint_(""),
             zmq_proto_(zmq_proto), zmq_context_(1),
             zmq_pub_sock_(zmq_context_, ZMQ_PUB),
-            zmq_message_ix(0)
+            zmq_message_ix(0),
+            m_allow_metadata(allow_metadata)
         { }
 
         DabOutputZMQ(const DabOutputZMQ& other) = delete;
@@ -338,6 +339,7 @@ class DabOutputZMQ : public DabOutput
         zmq_dab_message_t zmq_message;
         int zmq_message_ix;
 
+        bool m_allow_metadata;
         std::vector<std::shared_ptr<OutputMetadata> > meta_;
 };
 
diff --git a/src/dabOutput/dabOutputZMQ.cpp b/src/dabOutput/dabOutputZMQ.cpp
index 582af80..f50b7c3 100644
--- a/src/dabOutput/dabOutputZMQ.cpp
+++ b/src/dabOutput/dabOutputZMQ.cpp
@@ -126,7 +126,9 @@ int DabOutputZMQ::Close()
 
 void DabOutputZMQ::setMetadata(std::shared_ptr<OutputMetadata> &md)
 {
-    meta_.push_back(md);
+    if (m_allow_metadata) {
+        meta_.push_back(md);
+    }
 }
 
 #endif