summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/DabMux.cpp24
-rw-r--r--src/TcpSocket.cpp9
-rw-r--r--src/dabOutput/edi/Config.h6
-rw-r--r--src/dabOutput/edi/Transport.cpp20
-rw-r--r--src/dabOutput/edi/Transport.h1
-rw-r--r--src/zmq2edi/zmq2edi.cpp2
6 files changed, 52 insertions, 10 deletions
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;
generated by cgit v1.2.3 (git 2.46.0) at 2025-10-31 12:32:38 +0000