Update common EDI changes

6 files changed, 39 insertions, 19 deletions

diff --git a/contrib/Socket.cpp b/contrib/Socket.cpp
index 159de7e..bc1b179 100644
--- a/contrib/Socket.cpp
+++ b/contrib/Socket.cpp
@@ -2,7 +2,7 @@
    Copyright (C) 2003, 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the
    Queen in Right of Canada (Communications Research Center Canada)
 
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -932,6 +932,11 @@ void TCPSendClient::sendall(const std::vector<uint8_t>& buffer)
     }
 
     m_queue.push(buffer);
+
+    if (m_queue.size() > MAX_QUEUE_SIZE) {
+        vector<uint8_t> discard;
+        m_queue.try_pop(discard);
+    }
 }
 
 void TCPSendClient::process()
diff --git a/contrib/Socket.h b/contrib/Socket.h
index 84def40..8c6f8a9 100644
--- a/contrib/Socket.h
+++ b/contrib/Socket.h
@@ -2,7 +2,7 @@
    Copyright (C) 2003, 2004, 2005, 2006, 2007, 2008, 2009 Her Majesty the
    Queen in Right of Canada (Communications Research Center Canada)
 
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -311,7 +311,7 @@ class TCPSendClient {
         bool m_is_connected = false;
 
         TCPSocket m_sock;
-        static constexpr size_t MAX_QUEUE_SIZE = 1024;
+        static constexpr size_t MAX_QUEUE_SIZE = 512;
         ThreadsafeQueue<std::vector<uint8_t> > m_queue;
         std::atomic<bool> m_running;
         std::string m_exception_data;
diff --git a/contrib/edi/TagPacket.cpp b/contrib/edi/TagPacket.cpp
index b0bf9a1..ec52ad7 100644
--- a/contrib/edi/TagPacket.cpp
+++ b/contrib/edi/TagPacket.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2014
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -41,17 +41,19 @@ TagPacket::TagPacket(unsigned int alignment) : m_alignment(alignment)
 
 std::vector<uint8_t> TagPacket::Assemble()
 {
-    std::list<TagItem*>::iterator tag;
+    if (raw_tagpacket.size() > 0 and tag_items.size() > 0) {
+        throw std::logic_error("TagPacket: both raw and items used!");
+    }
 
-    std::vector<uint8_t> packet;
+    if (raw_tagpacket.size() > 0) {
+        return raw_tagpacket;
+    }
 
-    //std::cerr << "Assemble TAGPacket" << std::endl;
+    std::vector<uint8_t> packet;
 
-    for (tag = tag_items.begin(); tag != tag_items.end(); ++tag) {
-        std::vector<uint8_t> tag_data = (*tag)->Assemble();
+    for (auto tag : tag_items) {
+        std::vector<uint8_t> tag_data = tag->Assemble();
         packet.insert(packet.end(), tag_data.begin(), tag_data.end());
-
-        //std::cerr << "     Add TAGItem of length " << tag_data.size() << std::endl;
     }
 
     if (m_alignment == 0) { /* no padding */ }
diff --git a/contrib/edi/TagPacket.h b/contrib/edi/TagPacket.h
index 1e40ce7..b53b718 100644
--- a/contrib/edi/TagPacket.h
+++ b/contrib/edi/TagPacket.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2014
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -39,6 +39,9 @@ namespace edi {
 // Assemble function that puts the bytestream together and adds
 // padding such that the total length is a multiple of 8 Bytes.
 //
+// Alternatively, a raw tagpacket can be used instead of the
+// items list
+//
 // ETSI TS 102 821, 5.1 Tag Packet
 class TagPacket
 {
@@ -48,6 +51,8 @@ class TagPacket
 
         std::list<TagItem*> tag_items;
 
+        std::vector<uint8_t> raw_tagpacket;
+
     private:
         unsigned int m_alignment;
 };
diff --git a/contrib/edi/Transport.cpp b/contrib/edi/Transport.cpp
index 4c91483..d8627fd 100644
--- a/contrib/edi/Transport.cpp
+++ b/contrib/edi/Transport.cpp
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -33,7 +33,7 @@ namespace edi {
 
 void configuration_t::print() const
 {
-    etiLog.level(info) << "EDI";
+    etiLog.level(info) << "EDI Output";
     etiLog.level(info) << " verbose     " << verbose;
     for (auto edi_dest : destinations) {
         if (auto udp_dest = dynamic_pointer_cast<edi::udp_destination_t>(edi_dest)) {
@@ -67,7 +67,7 @@ Sender::Sender(const configuration_t& conf) :
     edi_pft(m_conf)
 {
     if (m_conf.verbose) {
-        etiLog.log(info, "Setup EDI");
+        etiLog.log(info, "Setup EDI Output");
     }
 
     for (const auto& edi_dest : m_conf.destinations) {
@@ -104,7 +104,7 @@ Sender::Sender(const configuration_t& conf) :
     }
 
     if (m_conf.verbose) {
-        etiLog.log(info, "EDI set up");
+        etiLog.log(info, "EDI output set up");
     }
 }
 
@@ -118,7 +118,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\n",
+            fprintf(stderr, "EDI Output: Number of PFT fragment before interleaver %zu\n",
                     edi_fragments.size());
         }
 
@@ -127,7 +127,7 @@ void Sender::write(const TagPacket& tagpacket)
         }
 
         if (m_conf.verbose) {
-            fprintf(stderr, "EDI number of PFT fragments %zu\n",
+            fprintf(stderr, "EDI Output: Number of PFT fragments %zu\n",
                     edi_fragments.size());
         }
 
@@ -169,6 +169,12 @@ void Sender::write(const TagPacket& tagpacket)
                 Socket::InetAddress addr;
                 addr.resolveUdpDestination(udp_dest->dest_addr, m_conf.dest_port);
 
+                if (af_packet.size() > 1400 and not m_udp_fragmentation_warning_printed) {
+                    fprintf(stderr, "EDI Output: AF packet larger than 1400,"
+                            " consider using PFT to avoid UP fragmentation.\n");
+                    m_udp_fragmentation_warning_printed = true;
+                }
+
                 udp_sockets.at(udp_dest.get())->send(af_packet, addr);
             }
             else if (auto tcp_dest = dynamic_pointer_cast<edi::tcp_server_t>(dest)) {
diff --git a/contrib/edi/Transport.h b/contrib/edi/Transport.h
index 73b2ab6..56ded3b 100644
--- a/contrib/edi/Transport.h
+++ b/contrib/edi/Transport.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (C) 2019
+   Copyright (C) 2020
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://www.opendigitalradio.org
@@ -50,6 +50,8 @@ class Sender {
         void write(const TagPacket& tagpacket);
 
     private:
+        bool m_udp_fragmentation_warning_printed = false;
+
         configuration_t m_conf;
         std::ofstream edi_debug_file;