allow FNE PUT /dmr/rid and /p25/rid to target *all* connected peers; …

…correct naming of packet data dumping configuration parameter; continue some work on P25 PDU data and VTUN; adjust P25 PDU ACK_RSP;

configs/fne-config.example.yml

@@ -69,7 +69,7 @@ master:
     # Flag indicating whether packet data will be passed.
     disablePacketData: false
     # Flag indicating whether verbose dumping of data packets is enabled.
-    dumpDataPacket: false
+    dumpPacketData: false
 
     # Delay from when a call on a parrot TG ends to when the playback starts (in milliseconds).
     parrotDelay: 2000

src/fne/HostFNE.cpp

@@ -206,6 +206,8 @@ int HostFNE::run()
 #if !defined(_WIN32)
     if (!Thread::runAsThread(this, threadVirtualNetworking))
         return EXIT_FAILURE;
+    if (!Thread::runAsThread(this, threadVirtualNetworkingClock))
+        return EXIT_FAILURE;
 #endif // !defined(_WIN32)
     /*
     ** Main execution loop
@@ -836,7 +838,7 @@ void* HostFNE::threadVirtualNetworking(void* arg)
     if (th != nullptr) {
         ::pthread_detach(th->thread);
 
-        std::string threadName("fne:vtun-loop");
+        std::string threadName("fne:vtun-net-rx");
         HostFNE* fne = static_cast<HostFNE*>(th->obj);
         if (fne == nullptr) {
             g_killed = true;
@@ -863,7 +865,6 @@ void* HostFNE::threadVirtualNetworking(void* arg)
             stopWatch.start();
 
             while (!g_killed) {
-                uint32_t ms = stopWatch.elapsed();
                 stopWatch.start();
 
                 uint8_t packet[DEFAULT_MTU_SIZE];
@@ -882,6 +883,55 @@ void* HostFNE::threadVirtualNetworking(void* arg)
                     }
                 }
 
+                Thread::sleep(2U);
+            }
+        }
+
+        LogDebug(LOG_HOST, "[STOP] %s", threadName.c_str());
+        delete th;
+    }
+
+    return nullptr;
+}
+
+/* Entry point to virtual networking clocking thread. */
+
+void* HostFNE::threadVirtualNetworkingClock(void* arg)
+{
+    thread_t* th = (thread_t*)arg;
+    if (th != nullptr) {
+        ::pthread_detach(th->thread);
+
+        std::string threadName("fne:vtun-clock");
+        HostFNE* fne = static_cast<HostFNE*>(th->obj);
+        if (fne == nullptr) {
+            g_killed = true;
+            LogDebug(LOG_HOST, "[FAIL] %s", threadName.c_str());
+        }
+
+        if (g_killed) {
+            delete th;
+            return nullptr;
+        }
+
+        if (!fne->m_vtunEnabled) {
+            delete th;
+            return nullptr;
+        }
+
+        LogDebug(LOG_HOST, "[ OK ] %s", threadName.c_str());
+#ifdef _GNU_SOURCE
+        ::pthread_setname_np(th->thread, threadName.c_str());
+#endif // _GNU_SOURCE
+
+        if (fne->m_tun != nullptr) {
+            StopWatch stopWatch;
+            stopWatch.start();
+
+            while (!g_killed) {
+                uint32_t ms = stopWatch.elapsed();
+                stopWatch.start();
+
                 // clock traffic handler
                 switch (fne->m_packetDataMode) {
                 case PacketDataMode::DMR:
@@ -893,7 +943,7 @@ void* HostFNE::threadVirtualNetworking(void* arg)
                     break;
                 }
 
-                Thread::sleep(5U);
+                Thread::sleep(2U);
             }
         }
 

src/fne/HostFNE.h

@@ -163,6 +163,12 @@ class HOST_SW_API HostFNE {
      * @returns void* (Ignore)
      */
     static void* threadVirtualNetworking(void* arg);
+    /**
+     * @brief Entry point to virtual networking clocking thread.
+     * @param arg Instance of the thread_t structure.
+     * @returns void* (Ignore)
+     */
+    static void* threadVirtualNetworkingClock(void* arg);
 #endif // !defined(_WIN32)
     /**
      * @brief Processes any peer network traffic.

src/fne/network/FNENetwork.cpp

@@ -94,7 +94,7 @@ FNENetwork::FNENetwork(HostFNE* host, const std::string& address, uint16_t port,
     m_influxBucket("dvm"),
     m_influxLogRawData(false),
     m_disablePacketData(false),
-    m_dumpDataPacket(false),
+    m_dumpPacketData(false),
     m_reportPeerPing(reportPeerPing),
     m_verbose(verbose)
 {
@@ -153,7 +153,7 @@ void FNENetwork::setOptions(yaml::Node& conf, bool printOptions)
     m_filterTerminators = conf["filterTerminators"].as<bool>(true);
 
     m_disablePacketData = conf["disablePacketData"].as<bool>(false);
-    m_dumpDataPacket = conf["dumpDataPacket"].as<bool>(false);
+    m_dumpPacketData = conf["dumpPacketData"].as<bool>(false);
 
     /*
     ** Drop Unit to Unit Peers
@@ -176,7 +176,7 @@ void FNENetwork::setOptions(yaml::Node& conf, bool printOptions)
             LogWarning(LOG_NET, "NOTICE: All P25 ADJ_STS_BCAST messages will be blocked and dropped!");
         }
         LogInfo("    Disable Packet Data: %s", m_disablePacketData ? "yes" : "no");
-        LogInfo("    Dump Packet Data: %s", m_dumpDataPacket ? "yes" : "no");
+        LogInfo("    Dump Packet Data: %s", m_dumpPacketData ? "yes" : "no");
         LogInfo("    Disable P25 ADJ_STS_BCAST to external peers: %s", m_disallowExtAdjStsBcast ? "yes" : "no");
         LogInfo("    Allow conventional sites to override affiliation and receive all traffic: %s", m_allowConvSiteAffOverride ? "yes" : "no");
         LogInfo("    Restrict grant response by affiliation: %s", m_restrictGrantToAffOnly ? "yes" : "no");

src/fne/network/FNENetwork.h

@@ -475,7 +475,7 @@ namespace network
         influxdb::ServerInfo m_influxServer;
 
         bool m_disablePacketData;
-        bool m_dumpDataPacket;
+        bool m_dumpPacketData;
 
         bool m_reportPeerPing;
         bool m_verbose;

src/fne/network/RESTAPI.cpp

@@ -1419,12 +1419,6 @@ void RESTAPI::restAPI_PutDMRRID(const HTTPPayload& request, HTTPPayload& reply,
         return;
     }
 
-    // validate peer ID is a integer within the JSON blob
-    if (!req["peerId"].is<uint32_t>()) {
-        errorPayload(reply, "peer ID was not valid");
-        return;
-    }
-
     // validate destination ID is a integer within the JSON blob
     if (!req["dstId"].is<uint32_t>()) {
         errorPayload(reply, "destination ID was not valid");
@@ -1437,15 +1431,10 @@ void RESTAPI::restAPI_PutDMRRID(const HTTPPayload& request, HTTPPayload& reply,
         return;
     }
 
-    uint32_t peerId = req["peerId"].get<uint32_t>();     
+    uint32_t peerId = req["peerId"].getDefault<uint32_t>(0U);
     uint32_t dstId = req["dstId"].get<uint32_t>();
     uint8_t slot = req["slot"].get<uint8_t>();
 
-    if (peerId == 0U) {
-        errorPayload(reply, "peer ID was not valid");
-        return;
-    }
-
     if (dstId == 0U) {
         errorPayload(reply, "destination ID was not valid");
         return;
@@ -1500,26 +1489,15 @@ void RESTAPI::restAPI_PutP25RID(const HTTPPayload& request, HTTPPayload& reply,
         return;
     }
 
-    // validate peer ID is a integer within the JSON blob
-    if (!req["peerId"].is<uint32_t>()) {
-        errorPayload(reply, "peer ID was not valid");
-        return;
-    }
-
     // validate destination ID is a integer within the JSON blob
     if (!req["dstId"].is<uint32_t>()) {
         errorPayload(reply, "destination ID was not valid");
         return;
     }
 
-    uint32_t peerId = req["peerId"].get<uint32_t>();     
+    uint32_t peerId = req["peerId"].getDefault<uint32_t>(0U);
     uint32_t dstId = req["dstId"].get<uint32_t>();
 
-    if (peerId == 0U) {
-        errorPayload(reply, "peer ID was not valid");
-        return;
-    }
-
     if (dstId == 0U) {
         errorPayload(reply, "destination ID was not valid");
         return;

src/fne/network/callhandler/TagDMRData.cpp

@@ -952,5 +952,39 @@ void TagDMRData::write_CSBK(uint32_t peerId, uint8_t slot, lc::CSBK* csbk)
         return;
     }
 
-    m_network->writePeer(peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId, false, true);
+    if (peerId > 0U) {
+        m_network->writePeer(peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId, false, true);
+    } else {
+        // repeat traffic to the connected peers
+        if (m_network->m_peers.size() > 0U) {
+            uint32_t i = 0U;
+            for (auto peer : m_network->m_peers) {
+                // every 5 peers flush the queue
+                if (i % 5U == 0U) {
+                    m_network->m_frameQueue->flushQueue();
+                }
+
+                m_network->writePeer(peer.first, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId, true);
+                if (m_network->m_debug) {
+                    LogDebug(LOG_NET, "DMR, peer = %u, slotNo = %u, len = %u, stream = %u", 
+                        peer.first, slot, messageLength, streamId);
+                }
+                i++;
+            }
+            m_network->m_frameQueue->flushQueue();
+        }
+
+        // repeat traffic to external peers
+        if (m_network->m_host->m_peerNetworks.size() > 0U) {
+            for (auto peer : m_network->m_host->m_peerNetworks) {
+                uint32_t dstPeerId = peer.second->getPeerId();
+
+                peer.second->writeMaster({ NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId);
+                if (m_network->m_debug) {
+                    LogDebug(LOG_NET, "DMR, peer = %u, slotNo = %u, len = %u, stream = %u", 
+                        dstPeerId, slot, messageLength, streamId);
+                }
+            }
+        }
+    }
 }

src/fne/network/callhandler/TagP25Data.cpp

@@ -1283,5 +1283,38 @@ void TagP25Data::write_TSDU(uint32_t peerId, lc::TSBK* tsbk)
     }
 
     uint32_t streamId = m_network->createStreamId();
-    m_network->writePeer(peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_P25 }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId, false, true);
+    if (peerId > 0U) {
+        m_network->writePeer(peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_P25 }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId, false, true);
+    } else {
+        // repeat traffic to the connected peers
+        if (m_network->m_peers.size() > 0U) {
+            uint32_t i = 0U;
+            for (auto peer : m_network->m_peers) {
+                // every 5 peers flush the queue
+                if (i % 5U == 0U) {
+                    m_network->m_frameQueue->flushQueue();
+                }
+
+                m_network->writePeer(peer.first, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_P25 }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId, true);
+                if (m_network->m_debug) {
+                    LogDebug(LOG_NET, "P25, peer = %u, len = %u, streamId = %u", 
+                        peer.first, messageLength, streamId);
+                }
+                i++;
+            }
+            m_network->m_frameQueue->flushQueue();
+        }
+
+        // repeat traffic to external peers
+        if (m_network->m_host->m_peerNetworks.size() > 0U) {
+            for (auto peer : m_network->m_host->m_peerNetworks) {
+                uint32_t dstPeerId = peer.second->getPeerId();
+                peer.second->writeMaster({ NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_P25 }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId);
+                if (m_network->m_debug) {
+                    LogDebug(LOG_NET, "P25, peer = %u, len = %u, streamId = %u", 
+                        dstPeerId, messageLength, streamId);
+                }
+            }
+        }
+    }
 }

src/fne/network/callhandler/packetdata/DMRPacketData.cpp

@@ -253,8 +253,8 @@ void DMRPacketData::dispatch(uint32_t peerId, dmr::data::NetData& dmrData, const
             LogWarning(LOG_NET, P25_PDU_STR ", failed CRC-32 check, blocks %u, len %u", status->header.getBlocksToFollow(), status->pduDataOffset);
         }
 
-        if (m_network->m_dumpDataPacket) {
-            Utils::dump(1U, "PDU Packet", status->pduUserData, status->pduDataOffset);
+        if (m_network->m_dumpPacketData) {
+            Utils::dump(1U, "ISP PDU Packet", status->pduUserData, status->pduDataOffset);
         }
     }
 }