diff options
| -rw-r--r-- | Makefile.am | 2 | ||||
| -rw-r--r-- | doc/advanced.mux | 18 | ||||
| -rw-r--r-- | src/DabMux.cpp | 24 | ||||
| -rw-r--r-- | src/TcpSocket.cpp | 9 | ||||
| -rw-r--r-- | src/dabOutput/edi/Config.h | 6 | ||||
| -rw-r--r-- | src/dabOutput/edi/Transport.cpp | 20 | ||||
| -rw-r--r-- | src/dabOutput/edi/Transport.h | 1 | ||||
| -rw-r--r-- | src/zmq2edi/zmq2edi.cpp | 2 |
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; |