Work on STI-D/EDI input

10 files changed, 81 insertions, 398 deletions

diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp
index fb49efc..3142bb3 100644
--- a/src/ConfigParser.cpp
+++ b/src/ConfigParser.cpp
@@ -40,6 +40,7 @@
 #include "utils.h"
 #include "DabMux.h"
 #include "ManagementServer.h"
+#include "input/Edi.h"
 #include "input/Prbs.h"
 #include "input/Zmq.h"
 #include "input/File.h"
@@ -876,34 +877,46 @@ static void setup_subchannel_from_ptree(shared_ptr<DabSubchannel>& subchan,
         type = pt.get<string>("type");
     }
     catch (const ptree_error &e) {
-        stringstream ss;
-        ss << "Subchannel with uid " << subchanuid << " has no type defined!";
-        throw runtime_error(ss.str());
+        throw runtime_error("Subchannel with uid " + subchanuid + " has no type defined!");
     }
 
-    /* Both inputfile and inputuri are supported, and are equivalent.
-     * inputuri has precedence
+    /* Up to v2.3.1, both inputfile and inputuri are supported, and are
+     * equivalent.  inputuri has precedence.
+     *
+     * After that, either inputfile or the (inputproto, inputuri) pair must be given, but not both.
      */
     string inputUri = pt.get<string>("inputuri", "");
+    string proto = pt.get<string>("inputproto", "");
 
-    if (inputUri == "") {
+    if (inputUri.empty() and proto.empty()) {
         try {
+            /* Old approach, derives proto from scheme used in the URL.
+             * This makes it impossible to distinguish between ZMQ tcp:// and
+             * EDI tcp://
+             */
             inputUri = pt.get<string>("inputfile");
+            size_t protopos = inputUri.find("://");
+
+            if (protopos == string::npos) {
+                proto = "file";
+            }
+            else {
+                proto = inputUri.substr(0, protopos);
+
+                if (proto == "tcp" or proto == "epgm" or proto == "ipc") {
+                    proto = "zmq";
+                }
+                else if (proto == "sti-rtp") {
+                    proto = "sti";
+                }
+            }
         }
         catch (const ptree_error &e) {
-            stringstream ss;
-            ss << "Subchannel with uid " << subchanuid << " has no inputUri defined!";
-            throw runtime_error(ss.str());
+            throw runtime_error("Subchannel with uid " + subchanuid + " has no input defined!");
         }
     }
-
-    string proto;
-    size_t protopos = inputUri.find("://");
-    if (protopos == string::npos) {
-        proto = "file";
-    }
-    else {
-        proto = inputUri.substr(0, protopos);
+    else if (inputUri.empty() or proto.empty()) {
+        throw runtime_error("Must define both inputuri and inputproto for uid " + subchanuid);
     }
 
     subchan->inputUri = inputUri;
@@ -928,7 +941,7 @@ static void setup_subchannel_from_ptree(shared_ptr<DabSubchannel>& subchan,
                 throw logic_error("Incomplete handling of file input");
             }
         }
-        else if (proto == "tcp"  or proto == "epgm" or proto == "ipc") {
+        else if (proto == "zmq") {
             auto zmqconfig = setup_zmq_input(pt, subchanuid);
 
             if (type == "audio") {
@@ -941,15 +954,11 @@ static void setup_subchannel_from_ptree(shared_ptr<DabSubchannel>& subchan,
                 rcs.enrol(inzmq.get());
                 subchan->input = inzmq;
             }
-
-            if (proto == "epgm") {
-                etiLog.level(warn) << "Using untested epgm:// zeromq input";
-            }
-            else if (proto == "ipc") {
-                etiLog.level(warn) << "Using untested ipc:// zeromq input";
-            }
         }
-        else if (proto == "sti-rtp") {
+        else if (proto == "edi") {
+            subchan->input = make_shared<Inputs::Edi>();
+        }
+        else if (proto == "stp") {
             subchan->input = make_shared<Inputs::Sti_d_Rtp>();
         }
         else {
diff --git a/src/DabMux.cpp b/src/DabMux.cpp
index d749ed3..e726fd3 100644
--- a/src/DabMux.cpp
+++ b/src/DabMux.cpp
@@ -304,7 +304,7 @@ int main(int argc, char *argv[])
                         edi_conf.destinations.push_back(dest);
                     }
                     else if (proto == "tcp") {
-                        auto dest = make_shared<edi::tcp_destination_t>();
+                        auto dest = make_shared<edi::tcp_server_t>();
                         dest->listen_port = pt_edi_dest.second.get<unsigned int>("listenport");
                         dest->max_frames_queued = pt_edi_dest.second.get<size_t>("max_frames_queued", 500);
                         edi_conf.destinations.push_back(dest);
diff --git a/src/ReedSolomon.cpp b/src/ReedSolomon.cpp
deleted file mode 100644
index 38d8ea8..0000000
--- a/src/ReedSolomon.cpp
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
-   Copyright (C) 2005, 2006, 2007, 2008, 2009 Her Majesty the Queen in Right
-   of Canada (Communications Research Center Canada)
-
-   Copyright (C) 2016
-   Matthias P. Braendli, matthias.braendli@mpb.li
-
-    http://www.opendigitalradio.org
-   */
-/*
-   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 "ReedSolomon.h"
-#include <vector>
-#include <algorithm>
-#include <stdexcept>
-#include <sstream>
-#include <stdio.h>          // For galois.h ...
-#include <string.h>         // For memcpy
-
-extern "C" {
-#include "fec/fec.h"
-}
-#include <assert.h>
-
-#define SYMSIZE     8
-
-
-ReedSolomon::ReedSolomon(int N, int K, bool reverse, int gfpoly, int firstRoot, int primElem)
-{
-    setReverse(reverse);
-
-    m_N = N;
-    m_K = K;
-
-    const int symsize = SYMSIZE;
-    const int nroots = N - K; // For EDI PFT, this must be 48
-    const int pad = ((1 << symsize) - 1) - N; // is 255-N
-
-    rsData = init_rs_char(symsize, gfpoly, firstRoot, primElem, nroots, pad);
-
-    if (rsData == nullptr) {
-        std::stringstream ss;
-        ss << "Invalid Reed-Solomon parameters! " <<
-            "N=" << N << " ; K=" << K << " ; pad=" << pad;
-        throw std::invalid_argument(ss.str());
-    }
-}
-
-
-ReedSolomon::~ReedSolomon()
-{
-    free_rs_char(rsData);
-}
-
-
-void ReedSolomon::setReverse(bool state)
-{
-    reverse = state;
-}
-
-
-int ReedSolomon::encode(void* data, void* fec, size_t size)
-{
-    uint8_t* input = reinterpret_cast<uint8_t*>(data);
-    uint8_t* output = reinterpret_cast<uint8_t*>(fec);
-    int ret = 0;
-
-    if (reverse) {
-        std::vector<uint8_t> buffer(m_N);
-
-        memcpy(&buffer[0], input, m_K);
-        memcpy(&buffer[m_K], output, m_N - m_K);
-
-        ret = decode_rs_char(rsData, &buffer[0], nullptr, 0);
-        if ((ret != 0) && (ret != -1)) {
-            memcpy(input, &buffer[0], m_K);
-            memcpy(output, &buffer[m_K], m_N - m_K);
-        }
-    }
-    else {
-        encode_rs_char(rsData, input, output);
-    }
-
-    return ret;
-}
-
-
-int ReedSolomon::encode(void* data, size_t size)
-{
-    uint8_t* input = reinterpret_cast<uint8_t*>(data);
-    int ret = 0;
-
-    if (reverse) {
-        ret = decode_rs_char(rsData, input, nullptr, 0);
-    }
-    else {
-        encode_rs_char(rsData, input, &input[m_K]);
-    }
-
-    return ret;
-}
diff --git a/src/ReedSolomon.h b/src/ReedSolomon.h
deleted file mode 100644
index abcef62..0000000
--- a/src/ReedSolomon.h
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
-   Copyright (C) 2005, 2006, 2007, 2008, 2009 Her Majesty the Queen in Right
-   of Canada (Communications Research Center Canada)
-
-   Copyright (C) 2016
-   Matthias P. Braendli, matthias.braendli@mpb.li
-
-    http://www.opendigitalradio.org
-   */
-/*
-   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
-
-#ifdef HAVE_CONFIG_H
-# include <config.h>
-#endif
-
-#include <stdlib.h>
-
-class ReedSolomon
-{
-public:
-    ReedSolomon(int N, int K,
-            bool reverse = false,
-            int gfpoly = 0x11d, int firstRoot = 0, int primElem = 1);
-    ReedSolomon(const ReedSolomon& other) = delete;
-    ReedSolomon operator=(const ReedSolomon& other) = delete;
-    ~ReedSolomon();
-
-    void setReverse(bool state);
-    int encode(void* data, void* fec, size_t size);
-    int encode(void* data, size_t size);
-
-private:
-    int m_N;
-    int m_K;
-
-    void* rsData;
-    bool reverse;
-};
-
diff --git a/src/ThreadsafeQueue.h b/src/ThreadsafeQueue.h
deleted file mode 100644
index ab287b2..0000000
--- a/src/ThreadsafeQueue.h
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
-   Copyright (C) 2007, 2008, 2009, 2010, 2011 Her Majesty the Queen in
-   Right of Canada (Communications Research Center Canada)
-
-   Copyright (C) 2018
-   Matthias P. Braendli, matthias.braendli@mpb.li
-
-   An implementation for a threadsafe queue, depends on C++11
-
-   When creating a ThreadsafeQueue, one can specify the minimal number
-   of elements it must contain before it is possible to take one
-   element out.
- */
-/*
-   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 <mutex>
-#include <condition_variable>
-#include <queue>
-#include <utility>
-
-/* This queue is meant to be used by two threads. One producer
- * that pushes elements into the queue, and one consumer that
- * retrieves the elements.
- *
- * The queue can make the consumer block until an element
- * is available, or a wakeup requested.
- */
-
-/* Class thrown by blocking pop to tell the consumer
- * that there's a wakeup requested. */
-class ThreadsafeQueueWakeup {};
-
-template<typename T>
-class ThreadsafeQueue
-{
-public:
-    /* Push one element into the queue, and notify another thread that
-     * might be waiting.
-     *
-     * returns the new queue size.
-     */
-    size_t push(T const& val)
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        the_queue.push(val);
-        size_t queue_size = the_queue.size();
-        lock.unlock();
-
-        the_rx_notification.notify_one();
-
-        return queue_size;
-    }
-
-    size_t push(T&& val)
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        the_queue.emplace(std::move(val));
-        size_t queue_size = the_queue.size();
-        lock.unlock();
-
-        the_rx_notification.notify_one();
-
-        return queue_size;
-    }
-
-    /* Push one element into the queue, but wait until the
-     * queue size goes below the threshold.
-     *
-     * Notify waiting thread.
-     *
-     * returns the new queue size.
-     */
-    size_t push_wait_if_full(T const& val, size_t threshold)
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        while (the_queue.size() >= threshold) {
-            the_tx_notification.wait(lock);
-        }
-        the_queue.push(val);
-        size_t queue_size = the_queue.size();
-        lock.unlock();
-
-        the_rx_notification.notify_one();
-
-        return queue_size;
-    }
-
-    /* Trigger a wakeup event on a blocking consumer, which
-     * will receive a ThreadsafeQueueWakeup exception.
-     */
-    void trigger_wakeup(void)
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        wakeup_requested = true;
-        lock.unlock();
-        the_rx_notification.notify_one();
-    }
-
-    /* Send a notification for the receiver thread */
-    void notify(void)
-    {
-        the_rx_notification.notify_one();
-    }
-
-    bool empty() const
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        return the_queue.empty();
-    }
-
-    size_t size() const
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        return the_queue.size();
-    }
-
-    bool try_pop(T& popped_value)
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        if (the_queue.empty()) {
-            return false;
-        }
-
-        popped_value = the_queue.front();
-        the_queue.pop();
-
-        lock.unlock();
-        the_tx_notification.notify_one();
-
-        return true;
-    }
-
-    void wait_and_pop(T& popped_value, size_t prebuffering = 1)
-    {
-        std::unique_lock<std::mutex> lock(the_mutex);
-        while (the_queue.size() < prebuffering and
-                not wakeup_requested) {
-            the_rx_notification.wait(lock);
-        }
-
-        if (wakeup_requested) {
-            wakeup_requested = false;
-            throw ThreadsafeQueueWakeup();
-        }
-        else {
-            std::swap(popped_value, the_queue.front());
-            the_queue.pop();
-
-            lock.unlock();
-            the_tx_notification.notify_one();
-        }
-    }
-
-private:
-    std::queue<T> the_queue;
-    mutable std::mutex the_mutex;
-    std::condition_variable the_rx_notification;
-    std::condition_variable the_tx_notification;
-    bool wakeup_requested = false;
-};
-
diff --git a/src/dabOutput/edi/Config.h b/src/dabOutput/edi/Config.h
index 55d5f0f..0c7dce8 100644
--- a/src/dabOutput/edi/Config.h
+++ b/src/dabOutput/edi/Config.h
@@ -50,11 +50,18 @@ struct udp_destination_t : public destination_t {
 };
 
 // TCP server that can accept multiple connections
-struct tcp_destination_t : public destination_t {
+struct tcp_server_t : public destination_t {
     unsigned int listen_port = 0;
     size_t max_frames_queued = 1024;
 };
 
+// TCP client that connects to one endpoint
+struct tcp_client_t : public destination_t {
+    std::string dest_addr;
+    unsigned int dest_port = 0;
+    size_t max_frames_queued = 1024;
+};
+
 struct configuration_t {
     unsigned chunk_len = 207;        // RSk, data length of each chunk
     unsigned fec       = 0;          // number of fragments that can be recovered
diff --git a/src/dabOutput/edi/PFT.cpp b/src/dabOutput/edi/PFT.cpp
index 5b93016..63dfa34 100644
--- a/src/dabOutput/edi/PFT.cpp
+++ b/src/dabOutput/edi/PFT.cpp
@@ -314,7 +314,7 @@ std::vector< PFTFragment > PFT::Assemble(AFPacket af_packet)
 
 #if 0
         fprintf(stderr, "* PFT pseq %d, findex %d, fcount %d, plen %d\n",
-                m_pseq, findex, fcount, plen & ~0x8000);
+                m_pseq, findex, fcount, plen & ~0xC000);
 #endif
     }
 
diff --git a/src/dabOutput/edi/Transport.cpp b/src/dabOutput/edi/Transport.cpp
index 6d3950f..187aabe 100644
--- a/src/dabOutput/edi/Transport.cpp
+++ b/src/dabOutput/edi/Transport.cpp
@@ -45,12 +45,16 @@ void configuration_t::print() const
             }
             etiLog.level(info) << "  source port " << udp_dest->source_port;
         }
-        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_destination_t>(edi_dest)) {
+        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(edi_dest)) {
             etiLog.level(info) << " TCP listening on port " << tcp_dest->listen_port;
             etiLog.level(info) << "  max frames queued    " << tcp_dest->max_frames_queued;
         }
+        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(edi_dest)) {
+            etiLog.level(info) << " TCP client connecting to " << tcp_dest->dest_addr << ":" << tcp_dest->dest_port;
+            etiLog.level(info) << "  max frames queued    " << tcp_dest->max_frames_queued;
+        }
         else {
-            throw std::logic_error("EDI destination not implemented");
+            throw logic_error("EDI destination not implemented");
         }
     }
     if (interleaver_enabled()) {
@@ -78,13 +82,18 @@ Sender::Sender(const configuration_t& conf) :
 
             udp_sockets.emplace(udp_dest.get(), udp_socket);
         }
-        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_destination_t>(edi_dest)) {
+        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(edi_dest)) {
             auto dispatcher = make_shared<Socket::TCPDataDispatcher>(tcp_dest->max_frames_queued);
             dispatcher->start(tcp_dest->listen_port, "0.0.0.0");
             tcp_dispatchers.emplace(tcp_dest.get(), dispatcher);
         }
+        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(edi_dest)) {
+            auto tcp_socket = make_shared<Socket::TCPSocket>();
+            tcp_socket->connect(tcp_dest->dest_addr, tcp_dest->dest_port);
+            tcp_senders.emplace(tcp_dest.get(), tcp_socket);
+        }
         else {
-            throw std::logic_error("EDI destination not implemented");
+            throw logic_error("EDI destination not implemented");
         }
     }
 
@@ -111,7 +120,7 @@ void Sender::write(const TagPacket& tagpacket)
         vector<edi::PFTFragment> edi_fragments = edi_pft.Assemble(af_packet);
 
         if (m_conf.verbose) {
-            fprintf(stderr, "EDI number of PFT fragment before interleaver %zu",
+            fprintf(stderr, "EDI number of PFT fragment before interleaver %zu\n",
                     edi_fragments.size());
         }
 
@@ -128,22 +137,25 @@ void Sender::write(const TagPacket& tagpacket)
 
                     udp_sockets.at(udp_dest.get())->send(edi_frag, addr);
                 }
-                else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_destination_t>(dest)) {
+                else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(dest)) {
                     tcp_dispatchers.at(tcp_dest.get())->write(edi_frag);
                 }
+                else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(dest)) {
+                    tcp_senders.at(tcp_dest.get())->sendall(edi_frag.data(), edi_frag.size());
+                }
                 else {
-                    throw std::logic_error("EDI destination not implemented");
+                    throw logic_error("EDI destination not implemented");
                 }
             }
 
             if (m_conf.dump) {
-                std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
-                std::copy(edi_frag.begin(), edi_frag.end(), debug_iterator);
+                ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
+                copy(edi_frag.begin(), edi_frag.end(), debug_iterator);
             }
         }
 
         if (m_conf.verbose) {
-            fprintf(stderr, "EDI number of PFT fragments %zu",
+            fprintf(stderr, "EDI number of PFT fragments %zu\n",
                     edi_fragments.size());
         }
     }
@@ -156,17 +168,20 @@ void Sender::write(const TagPacket& tagpacket)
 
                 udp_sockets.at(udp_dest.get())->send(af_packet, addr);
             }
-            else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_destination_t>(dest)) {
+            else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(dest)) {
                 tcp_dispatchers.at(tcp_dest.get())->write(af_packet);
             }
+            else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_client_t>(dest)) {
+                tcp_senders.at(tcp_dest.get())->sendall(af_packet.data(), af_packet.size());
+            }
             else {
-                throw std::logic_error("EDI destination not implemented");
+                throw logic_error("EDI destination not implemented");
             }
         }
 
         if (m_conf.dump) {
-            std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
-            std::copy(af_packet.begin(), af_packet.end(), debug_iterator);
+            ostream_iterator<uint8_t> debug_iterator(edi_debug_file);
+            copy(af_packet.begin(), af_packet.end(), debug_iterator);
         }
     }
 }
diff --git a/src/dabOutput/edi/Transport.h b/src/dabOutput/edi/Transport.h
index 74126d1..9633275 100644
--- a/src/dabOutput/edi/Transport.h
+++ b/src/dabOutput/edi/Transport.h
@@ -36,6 +36,7 @@
 #include <vector>
 #include <unordered_map>
 #include <stdexcept>
+#include <fstream>
 #include <cstdint>
 
 namespace edi {
@@ -62,7 +63,8 @@ class Sender {
         edi::Interleaver edi_interleaver;
 
         std::unordered_map<udp_destination_t*, std::shared_ptr<Socket::UDPSocket>> udp_sockets;
-        std::unordered_map<tcp_destination_t*, std::shared_ptr<Socket::TCPDataDispatcher>> tcp_dispatchers;
+        std::unordered_map<tcp_server_t*, std::shared_ptr<Socket::TCPDataDispatcher>> tcp_dispatchers;
+        std::unordered_map<tcp_client_t*, std::shared_ptr<Socket::TCPSocket>> tcp_senders;
 };
 
 }
diff --git a/src/input/Udp.cpp b/src/input/Udp.cpp
index b4cced0..5d4f964 100644
--- a/src/input/Udp.cpp
+++ b/src/input/Udp.cpp
@@ -151,8 +151,8 @@ static uint16_t unpack2(const uint8_t *buf)
 
 int Sti_d_Rtp::open(const std::string& name)
 {
-    // Skip the sti-rtp:// part if it is present
-    const string endpoint = (name.substr(0, 10) == "sti-rtp://") ?
+    // Skip the rtp:// part if it is present
+    const string endpoint = (name.substr(0, 10) == "rtp://") ?
         name.substr(10) : name;
 
     // The endpoint should be address:port
@@ -160,8 +160,8 @@ int Sti_d_Rtp::open(const std::string& name)
     if (colon_pos == string::npos) {
         stringstream ss;
         ss << "'" << name <<
-                " is an invalid format for sti-rtp address: "
-                "expected [sti-rtp://]address:port";
+                " is an invalid format for rtp address: "
+                "expected [rtp://]address:port";
         throw invalid_argument(ss.str());
     }