Add edi time and utco as ZMQ metadata

Since ODR-DabMod doesn't check the zmq message length, it will accept
additional metadata at the end of the ETI data

6 files changed, 197 insertions, 11 deletions

diff --git a/src/DabMultiplexer.cpp b/src/DabMultiplexer.cpp
index 3421338..58a1c8f 100644
--- a/src/DabMultiplexer.cpp
+++ b/src/DabMultiplexer.cpp
@@ -658,9 +658,19 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
             edi_tagDETI.set_seconds(edi_time);
 
             // In case get_offset fails, we still want to update the EDI seconds
-            edi_tagDETI.set_tai_utc_offset(m_clock_tai.get_offset());
-        }
+            const auto utco = m_clock_tai.get_offset();
+            edi_tagDETI.set_tai_utc_offset(utco);
+
+            for (auto output : outputs) {
+                shared_ptr<OutputMetadata> md_utco =
+                    make_shared<OutputMetadataUTCO>(utco);
+                output->setMetadata(md_utco);
 
+                shared_ptr<OutputMetadata> md_edi_time =
+                    make_shared<OutputMetadataEDITime>(edi_tagDETI.seconds);
+                output->setMetadata(md_edi_time);
+            }
+        }
     }
     catch (std::runtime_error& e) {
         etiLog.level(error) << "Could not get UTC-TAI offset for EDI timestamp";
@@ -692,8 +702,6 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs
         edi_time += chrono::seconds(1);
     }
 
-
-
     /********************************************************************** 
      ***********   Section FRPD   *****************************************
      **********************************************************************/
diff --git a/src/Makefile.am b/src/Makefile.am
index 12934f0..557621c 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -60,6 +60,7 @@ odr_dabmux_SOURCES  =DabMux.cpp DabMux.h \
 					 dabOutput/dabOutputTcp.cpp \
 					 dabOutput/dabOutputUdp.cpp \
 					 dabOutput/dabOutputZMQ.cpp \
+					 dabOutput/metadata.h dabOutput/metadata.cpp \
 					 dabOutput/edi/AFPacket.cpp dabOutput/edi/AFPacket.h \
 					 dabOutput/edi/TagItems.cpp dabOutput/edi/TagItems.h \
 					 dabOutput/edi/TagPacket.cpp dabOutput/edi/TagPacket.h \
diff --git a/src/dabOutput/dabOutput.h b/src/dabOutput/dabOutput.h
index 2e5f489..58a2929 100644
--- a/src/dabOutput/dabOutput.h
+++ b/src/dabOutput/dabOutput.h
@@ -45,6 +45,7 @@
 #ifdef HAVE_OUTPUT_ZEROMQ
 #  include "zmq.hpp"
 #endif
+#include "dabOutput/metadata.h"
 
 /** Configuration for EDI output */
 
@@ -89,6 +90,8 @@ class DabOutput
         virtual ~DabOutput() {}
 
         virtual std::string get_info() const = 0;
+
+        virtual void setMetadata(std::shared_ptr<OutputMetadata> &md) = 0;
 };
 
 // ----- used in File and Fifo outputs
@@ -117,6 +120,7 @@ class DabOutputFile : public DabOutput
             return "file://" + filename_;
         }
 
+        virtual void setMetadata(std::shared_ptr<OutputMetadata> &md) {}
     protected:
         /* Set ETI type according to filename, and return
          * filename without the &type=foo part
@@ -176,6 +180,7 @@ class DabOutputRaw : public DabOutput
         std::string get_info() const {
             return "raw://" + filename_;
         }
+        virtual void setMetadata(std::shared_ptr<OutputMetadata> &md) {}
     private:
         std::string filename_;
         int socket_;
@@ -204,6 +209,7 @@ class DabOutputUdp : public DabOutput
         std::string get_info() const {
             return "udp://" + uri_;
         }
+        virtual void setMetadata(std::shared_ptr<OutputMetadata> &md) {}
     private:
         // make sure we don't copy this output around
         // the UdpPacket and UdpSocket do not support
@@ -228,7 +234,7 @@ class DabOutputTcp : public DabOutput
         std::string get_info() const {
             return "tcp://" + uri_;
         }
-
+        virtual void setMetadata(std::shared_ptr<OutputMetadata> &md) {}
     private:
         std::string uri_;
 
@@ -246,6 +252,7 @@ class DabOutputSimul : public DabOutput
         std::string get_info() const {
             return "simul://" + name_;
         }
+        virtual void setMetadata(std::shared_ptr<OutputMetadata> &md) {}
     private:
         std::string name_;
         std::chrono::steady_clock::time_point startTime_;
@@ -286,6 +293,10 @@ struct zmq_dab_message_t
      */
 
     uint8_t  buf[NUM_FRAMES_PER_ZMQ_MESSAGE*6144];
+
+    /* The packet is then followed with metadata appended to it,
+     * according to dabOutput/metadata.h
+     */
 };
 
 #define ZMQ_DAB_MESSAGE_HEAD_LENGTH (4 + NUM_FRAMES_PER_ZMQ_MESSAGE*2)
@@ -316,8 +327,9 @@ class DabOutputZMQ : public DabOutput
         int Open(const char* endpoint);
         int Write(void* buffer, int size);
         int Close();
-    private:
+        void setMetadata(std::shared_ptr<OutputMetadata> &md);
 
+    private:
         std::string endpoint_;
         std::string zmq_proto_;
         zmq::context_t zmq_context_; // handle for the zmq context
@@ -325,6 +337,8 @@ class DabOutputZMQ : public DabOutput
 
         zmq_dab_message_t zmq_message;
         int zmq_message_ix;
+
+        std::vector<std::shared_ptr<OutputMetadata> > meta_;
 };
 
 #endif
diff --git a/src/dabOutput/dabOutputZMQ.cpp b/src/dabOutput/dabOutputZMQ.cpp
index 6a9d669..582af80 100644
--- a/src/dabOutput/dabOutputZMQ.cpp
+++ b/src/dabOutput/dabOutputZMQ.cpp
@@ -88,19 +88,28 @@ int DabOutputZMQ::Write(void* buffer, int size)
     if (zmq_message_ix == NUM_FRAMES_PER_ZMQ_MESSAGE) {
 
         // Size of the header:
-        int full_length = ZMQ_DAB_MESSAGE_HEAD_LENGTH;
+        size_t full_length = ZMQ_DAB_MESSAGE_HEAD_LENGTH;
 
         for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
             full_length += zmq_message.buflen[i];
         }
 
-        zmq_message_ix = 0;
+        vector<uint8_t> msg(full_length);
+        memcpy(msg.data(), (uint8_t*)&zmq_message, full_length);
+
+        // metadata gets appended at the end
+        for (const auto& md : meta_) {
+            vector<uint8_t> md_data(md->getLength());
+            md->write(md_data.data());
+
+            copy(md_data.begin(), md_data.end(), back_inserter(msg));
+        }
 
         const int flags = 0;
-        zmq_send(zmq_pub_sock_,
-                (uint8_t*)&zmq_message,
-                full_length, flags);
+        zmq_send(zmq_pub_sock_, msg.data(), msg.size(), flags);
 
+        meta_.clear();
+        zmq_message_ix = 0;
         for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
             zmq_message.buflen[i] = -1;
         }
@@ -115,5 +124,10 @@ int DabOutputZMQ::Close()
     return zmq_close(zmq_pub_sock_);
 }
 
+void DabOutputZMQ::setMetadata(std::shared_ptr<OutputMetadata> &md)
+{
+    meta_.push_back(md);
+}
+
 #endif
 
diff --git a/src/dabOutput/metadata.cpp b/src/dabOutput/metadata.cpp
new file mode 100644
index 0000000..6dfd65c
--- /dev/null
+++ b/src/dabOutput/metadata.cpp
@@ -0,0 +1,70 @@
+/*
+   Copyright (C) 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the Queen in
+   Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+
+   The metadata support for the outputs.
+*/
+/*
+   This file is part of ODR-DabMux.
+
+   ODR-DabMux is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   ODR-DabMux is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
+   */
+
+#include "dabOutput/metadata.h"
+#include <cstring>
+#include <arpa/inet.h>
+
+#ifdef HAVE_CONFIG_H
+#   include "config.h"
+#endif
+
+template <typename T>
+size_t write_meta(output_metadata_id_e md, uint8_t *buf, T value)
+{
+    buf[0] = static_cast<uint8_t>(md);
+
+    const int16_t len_value = sizeof(T);
+
+    const uint16_t data_length = htons(len_value);
+    memcpy(buf + 1, &data_length, sizeof(data_length));
+
+    if (len_value == 1) {
+        buf[3] = value;
+    }
+    else if (len_value == 2) {
+        const uint16_t value = htons(value);
+        memcpy(buf + 3, &value, sizeof(value));
+    }
+    else if (len_value == 4) {
+        const uint32_t value = htons(value);
+        memcpy(buf + 3, &value, sizeof(value));
+    }
+
+    return 3 + len_value;
+}
+
+size_t OutputMetadataUTCO::write(uint8_t *buf)
+{
+    return write_meta(getId(), buf, utco);
+}
+
+size_t OutputMetadataEDITime::write(uint8_t *buf)
+{
+    return write_meta(getId(), buf, seconds);
+}
diff --git a/src/dabOutput/metadata.h b/src/dabOutput/metadata.h
new file mode 100644
index 0000000..e250938
--- /dev/null
+++ b/src/dabOutput/metadata.h
@@ -0,0 +1,79 @@
+/*
+   Copyright (C) 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the Queen in
+   Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://www.opendigitalradio.org
+
+   The metadata support for the outputs.
+*/
+/*
+   This file is part of ODR-DabMux.
+
+   ODR-DabMux is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   ODR-DabMux is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with ODR-DabMux.  If not, see <http://www.gnu.org/licenses/>.
+   */
+
+#pragma once
+#include "Log.h"
+
+/* Some outputs support additional metadata to be carried
+ * next to the main ETI stream. This metadata always has the
+ * following format:
+ *   Field:  | id | len      | value               |
+ *   Length:   1    2          depending on id
+ *
+ * Multi-byte values are transmitted in network byte order.
+ */
+
+enum class output_metadata_id_e {
+    // TAI-UTC offset, value is int16_t.
+    utc_offset = 1,
+
+    /* EDI Time is the number of SI seconds since 2000-01-01 T 00:00:00 UTC.
+     * value is an uint32_t */
+    edi_time = 2,
+};
+
+struct OutputMetadata {
+    virtual output_metadata_id_e getId(void) const = 0;
+    virtual size_t getLength(void) const = 0;
+
+    /* Write the value in the metadata format to the buffer.
+     * Returns number of bytes written. The buffer buf needs
+     * to be large enough to contain the value;
+     */
+    virtual size_t write(uint8_t *buf) = 0;
+};
+
+struct OutputMetadataUTCO : public OutputMetadata {
+    explicit OutputMetadataUTCO(int16_t utco) : utco(utco) {}
+    output_metadata_id_e getId(void) const { return output_metadata_id_e::utc_offset; }
+    virtual size_t getLength(void) const { return 5; }
+    virtual size_t write(uint8_t *buf);
+
+    int16_t utco;
+};
+
+struct OutputMetadataEDITime : public OutputMetadata {
+    explicit OutputMetadataEDITime(uint32_t seconds) : seconds(seconds) {}
+    output_metadata_id_e getId(void) const { return output_metadata_id_e::edi_time; }
+    virtual size_t getLength(void) const { return 7; }
+    virtual size_t write(uint8_t *buf);
+
+    uint32_t seconds;
+
+};
+