EDI: Implement TCP output

8 files changed, 70 insertions, 12 deletions

diff --git a/Makefile.am b/Makefile.am
index b756385..2f06879 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -207,6 +207,8 @@ odr_zmq2edi_SOURCES  = src/zmq2edi/zmq2edi.cpp \
 					   src/dabOutput/edi/Transport.h \
 					   src/InetAddress.h \
 					   src/InetAddress.cpp \
+					   src/TcpSocket.h \
+					   src/TcpSocket.cpp \
 					   src/UdpSocket.h \
 					   src/UdpSocket.cpp \
 					   src/ReedSolomon.h \
diff --git a/doc/advanced.mux b/doc/advanced.mux
index c2a4411..fb67b82 100644
--- a/doc/advanced.mux
+++ b/doc/advanced.mux
@@ -388,12 +388,14 @@ outputs {
             example_unicast {
                 ; example for unicast EDI over UDP
                 ; for unicast EDI, do not set source
+                protocol udp
                 destination "192.168.23.23"
                 sourceport  13000
             }
             example_multicast {
                 ; example for multicast EDI, the source IP is required
                 ; so that the data is sent on the correct ethernet interface
+                protocol udp
                 destination "232.20.10.1"
                 source      "192.168.0.50"
                 ; The multicast TTL has to be adapted according to your network
@@ -401,8 +403,20 @@ outputs {
 
                 sourceport  13000
             }
-
-            ; EDI over TCP is not supported
+            example_tcp {
+                ; example for EDI TCP server. TCP is reliable, so it is counterproductive to
+                ; use FEC. Using PFT also brings no benefit.
+                protocol tcp
+                listenport 13000
+
+                ; For every connected endpoint, a queue is created. If the queue overflows, we
+                ; assume the endpoint has a problem, and we close the connection. This sets
+                ; the max queue size in number of frames. With PFT disabled, one frame is generated
+                ; every 24ms. With PFT enabled, it depends on fragmentation and FEC settings.
+                ;
+                ; default value: 500 frames, without PFT: 12s worth of EDI data
+                ;max_frames_queued 500
+            }
         }
 
         ; The settings below apply to all destinations
diff --git a/src/DabMux.cpp b/src/DabMux.cpp
index 3d0a7d9..b9ee9fd 100644
--- a/src/DabMux.cpp
+++ b/src/DabMux.cpp
@@ -293,14 +293,26 @@ int main(int argc, char *argv[])
             if (outputuid == "edi") {
                 ptree pt_edi = pt_outputs.get_child("edi");
                 for (auto pt_edi_dest : pt_edi.get_child("destinations")) {
-                    auto dest = make_shared<edi::udp_destination_t>();
-                    dest->dest_addr   = pt_edi_dest.second.get<string>("destination");
-                    dest->ttl         = pt_edi_dest.second.get<unsigned int>("ttl", 1);
+                    const auto proto = pt_edi_dest.second.get<string>("protocol");
+                    if (proto == "udp") {
+                        auto dest = make_shared<edi::udp_destination_t>();
+                        dest->dest_addr   = pt_edi_dest.second.get<string>("destination");
+                        dest->ttl         = pt_edi_dest.second.get<unsigned int>("ttl", 1);
 
-                    dest->source_addr = pt_edi_dest.second.get<string>("source", "");
-                    dest->source_port = pt_edi_dest.second.get<unsigned int>("sourceport");
+                        dest->source_addr = pt_edi_dest.second.get<string>("source", "");
+                        dest->source_port = pt_edi_dest.second.get<unsigned int>("sourceport");
 
-                    edi_conf.destinations.push_back(dest);
+                        edi_conf.destinations.push_back(dest);
+                    }
+                    else if (proto == "tcp") {
+                        auto dest = make_shared<edi::tcp_destination_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);
+                    }
+                    else {
+                        throw runtime_error("Unknown EDI protocol " + proto);
+                    }
                 }
 
                 edi_conf.dest_port           = pt_edi.get<unsigned int>("port");
diff --git a/src/TcpSocket.cpp b/src/TcpSocket.cpp
index c05eace..3ebe73c 100644
--- a/src/TcpSocket.cpp
+++ b/src/TcpSocket.cpp
@@ -248,8 +248,10 @@ TCPConnection::TCPConnection(TcpSocket&& sock) :
             m_sender_thread(),
             m_sock(move(sock))
 {
+    auto own_addr = m_sock.getOwnAddress();
     auto addr = m_sock.getRemoteAddress();
-    etiLog.level(debug) << "New TCP Connection from " <<
+    etiLog.level(debug) << "New TCP Connection on port " <<
+        own_addr.getPort() << " from " <<
         addr.getHostAddress() << ":" << addr.getPort();
     m_sender_thread = std::thread(&TCPConnection::process, this);
 }
@@ -293,8 +295,11 @@ void TCPConnection::process()
         }
     }
 
+
+    auto own_addr = m_sock.getOwnAddress();
     auto addr = m_sock.getRemoteAddress();
-    etiLog.level(debug) << "Dropping TCP Connection from " <<
+    etiLog.level(debug) << "Dropping TCP Connection on port " <<
+        own_addr.getPort() << " from " <<
         addr.getHostAddress() << ":" << addr.getPort();
 }
 
diff --git a/src/dabOutput/edi/Config.h b/src/dabOutput/edi/Config.h
index d3678d9..55d5f0f 100644
--- a/src/dabOutput/edi/Config.h
+++ b/src/dabOutput/edi/Config.h
@@ -49,6 +49,12 @@ struct udp_destination_t : public destination_t {
     unsigned int ttl = 10;
 };
 
+// TCP server that can accept multiple connections
+struct tcp_destination_t : public destination_t {
+    unsigned int listen_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/Transport.cpp b/src/dabOutput/edi/Transport.cpp
index d433239..d99e987 100644
--- a/src/dabOutput/edi/Transport.cpp
+++ b/src/dabOutput/edi/Transport.cpp
@@ -38,13 +38,17 @@ void configuration_t::print() const
     etiLog.level(info) << " verbose     " << verbose;
     for (auto edi_dest : destinations) {
         if (auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) {
-            etiLog.level(info) << " to " << udp_dest->dest_addr << ":" << dest_port;
+            etiLog.level(info) << " UDP to " << udp_dest->dest_addr << ":" << dest_port;
             if (not udp_dest->source_addr.empty()) {
                 etiLog.level(info) << "  source      " << udp_dest->source_addr;
                 etiLog.level(info) << "  ttl         " << udp_dest->ttl;
             }
             etiLog.level(info) << "  source port " << udp_dest->source_port;
         }
+        else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_destination_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 {
             throw std::logic_error("EDI destination not implemented");
         }
@@ -80,6 +84,14 @@ 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)) {
+            auto dispatcher = make_shared<TCPDataDispatcher>(tcp_dest->max_frames_queued);
+            dispatcher->start(tcp_dest->listen_port, "0.0.0.0");
+            tcp_dispatchers.emplace(tcp_dest.get(), dispatcher);
+        }
+        else {
+            throw std::logic_error("EDI destination not implemented");
+        }
     }
 
     if (m_conf.interleaver_enabled()) {
@@ -123,6 +135,9 @@ 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)) {
+                    tcp_dispatchers.at(tcp_dest.get())->write(edi_frag);
+                }
                 else {
                     throw std::logic_error("EDI destination not implemented");
                 }
@@ -149,6 +164,9 @@ 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)) {
+                tcp_dispatchers.at(tcp_dest.get())->write(af_packet);
+            }
             else {
                 throw std::logic_error("EDI destination not implemented");
             }
diff --git a/src/dabOutput/edi/Transport.h b/src/dabOutput/edi/Transport.h
index 3c48c96..7b0a0db 100644
--- a/src/dabOutput/edi/Transport.h
+++ b/src/dabOutput/edi/Transport.h
@@ -62,6 +62,7 @@ class Sender {
         edi::Interleaver edi_interleaver;
 
         std::unordered_map<udp_destination_t*, std::shared_ptr<UdpSocket>> udp_sockets;
+        std::unordered_map<tcp_destination_t*, std::shared_ptr<TCPDataDispatcher>> tcp_dispatchers;
 };
 
 }
diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp
index ee5776e..a2daf49 100644
--- a/src/zmq2edi/zmq2edi.cpp
+++ b/src/zmq2edi/zmq2edi.cpp
@@ -64,7 +64,7 @@ void usage(void)
     cerr << " -v Enables verbose mode." << endl;
     cerr << " -a <tagpacket alignement> sets the alignment of the TAG Packet (default 8)." << endl << endl;
 
-    cerr << "The following options can be given several times, when more than once destination is addressed:" << endl;
+    cerr << "The following options can be given several times, when more than UDP destination is desired:" << endl;
     cerr << " -d <destination ip> sets the destination ip." << endl;
     cerr << " -s <source port> sets the source port." << endl;
     cerr << " -S <source ip> select the source IP in case we want to use multicast." << endl;