add support to store subscriber emerg alarm and calls in the operation store;

src/fne/network/SubscriberOperationStore.cpp

@@ -94,6 +94,34 @@ bool SubscriberOperationStore::complete(Protocol protocol, Kind kind, uint32_t t
     return false;
 }
 
+/* Adds a uniquely correlatable one-shot operation. */
+
+uint64_t SubscriberOperationStore::oneShot(Protocol protocol, Kind kind, uint32_t targetId, uint32_t peerId, uint16_t requestValue)
+{
+    const auto now = std::chrono::steady_clock::now();
+    std::lock_guard<std::mutex> lock(m_mutex);
+    expireLocked(now);
+
+    // suppress an equivalent one-shot while its result remains available
+    for (const auto& entry : m_operations) {
+        const Operation& operation = entry.second.value;
+        if (operation.protocol == protocol && operation.kind == kind && operation.targetId == targetId) {
+            return 0U;
+        }
+    }
+
+    // create a new operation with a unique ID and store it with expiration times
+    uint64_t id = m_nextId++;
+    if (id == 0U) {
+        id = m_nextId++;
+    }
+
+    const uint64_t createdAt = unixMilliseconds();
+    Operation operation { id, protocol, kind, State::SUCCEEDED, targetId, peerId, requestValue, 0U, createdAt, createdAt };
+    m_operations.emplace(id, StoredOperation { operation, now + OPERATION_TIMEOUT, now + RESULT_RETENTION });
+    return id;
+}
+
 /* Adds a status report and completes a matching P25 status query. */
 
 bool SubscriberOperationStore::addP25Status(uint32_t subscriberId, uint32_t peerId, uint16_t value)
@@ -192,7 +220,16 @@ void SubscriberOperationStore::expire()
 
 const char* SubscriberOperationStore::protocolName(Protocol protocol)
 {
-    return protocol == Protocol::DMR ? "dmr" : "p25";
+    switch (protocol) {
+    case Protocol::DMR:
+        return "dmr";
+    case Protocol::P25:
+        return "p25";
+    case Protocol::NXDN:
+        return "nxdn";
+    }
+
+    return "unknown";
 }
 
 /* Returns a stable operation name. */
@@ -216,6 +253,10 @@ const char* SubscriberOperationStore::kindName(Kind kind)
         return "kmm-changeover";
     case Kind::KMM_CHANGERSI:
         return "kmm-changersi";
+    case Kind::EMERG_ALRM:
+        return "emerg-alrm";
+    case Kind::EMERG_CALL:
+        return "emerg-call";
     }
 
     return "unknown";

src/fne/network/SubscriberOperationStore.h

@@ -42,7 +42,8 @@ namespace network
          */
         enum class Protocol : uint8_t {
             DMR = 0U,               //!< Digital Mobile Radio.
-            P25 = 1U                //!< Project 25.
+            P25 = 1U,               //!< Project 25.
+            NXDN = 2U               //!< Next Generation Digital Narrowband.
         };
 
         /**
@@ -58,7 +59,10 @@ namespace network
 
             KMM_REKEY = 5U,         //!< Key management rekey request.
             KMM_CHANGEOVER = 6U,    //!< Key management changeover request.
-            KMM_CHANGERSI = 7U      //!< Key management change RSI request.
+            KMM_CHANGERSI = 7U,     //!< Key management change RSI request.
+
+            EMERG_ALRM = 8U,        //!< Emergency alarm request.
+            EMERG_CALL = 9U         //!< Emergency voice call.
         };
 
         /**
@@ -130,6 +134,17 @@ namespace network
         bool complete(Protocol protocol, Kind kind, uint32_t targetId, uint32_t peerId,
             uint16_t resultValue = 0U, bool success = true);
 
+        /**
+         * @brief Adds a uniquely correlatable one-shot operation.
+         * @param[in] protocol Air-interface protocol.
+         * @param[in] kind Operation kind.
+         * @param[in] targetId Target subscriber ID.
+         * @param[in] peerId Requested peer ID, or zero for routed fanout.
+         * @param[in] requestValue Optional requested value.
+         * @returns Operation identifier, or zero when an equivalent request is already pending.
+         */
+        uint64_t oneShot(Protocol protocol, Kind kind, uint32_t targetId, uint32_t peerId, uint16_t requestValue = 0U);
+
         /**
          * @brief Adds a status report and completes a matching P25 status query.
          * @param[in] subscriberId Reporting subscriber ID.

src/fne/network/callhandler/TagDMRData.cpp

@@ -145,6 +145,12 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
         csbk = lc::csbk::CSBKFactory::createCSBK(frame, dataType);
     }
 
+    std::unique_ptr<lc::LC> voiceLC;
+    if (dataSync && dataType == DataType::VOICE_LC_HEADER) {
+        lc::FullLC fullLC;
+        voiceLC = fullLC.decode(frame, dataType);
+    }
+
     // perform TGID route rewrites if configured
     routeRewrite(buffer, peerId, dmrData, dataType, dstId, slotNo, false);
     dstId = GET_UINT24(buffer, 8U);
@@ -479,6 +485,11 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
                 m_status[dstId].activeCall = true;
                 m_status.unlock();
 
+                if (voiceLC != nullptr && voiceLC->getEmergency()) {
+                    m_network->subscriberOperations().oneShot(SubscriberOperationStore::Protocol::DMR,
+                        SubscriberOperationStore::Kind::EMERG_CALL, srcId, peerId);
+                }
+
                 // clear any rejected call streams for this TG
                 m_rejectedCallStreams.lock(false);
                 m_rejectedCallStreams[dstId].clear();

src/fne/network/callhandler/TagNXDNData.cpp

@@ -497,6 +497,11 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI
                     m_status[dstId].activeCall = true;
                     m_status.unlock();
 
+                    if (lc.getEmergency()) {
+                        m_network->subscriberOperations().oneShot(SubscriberOperationStore::Protocol::NXDN,
+                            SubscriberOperationStore::Kind::EMERG_CALL, srcId, peerId);
+                    }
+
                     // clear any rejected call streams for this TG
                     m_rejectedCallStreams.lock(false);
                     m_rejectedCallStreams[dstId].clear();

src/fne/network/callhandler/TagP25Data.cpp

@@ -201,6 +201,10 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
             LogWarning(LOG_NET, (duid == DUID::LDU1) ? P25_LDU1_STR : P25_LDU2_STR ", missing %u LDU voice frames, srcId = %u, dstId = %u", missing, srcId, dstId);
         }
 
+        if (duid == DUID::LDU1) {
+            control.setEmergency(dfsiLC.control()->getEmergency());
+        }
+
         // if its a LDU2 get the crypto state
         if (duid == DUID::LDU2) {
             control.setAlgId(dfsiLC.control()->getAlgId());
@@ -587,6 +591,11 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
                     m_status[dstId].activeCall = true;
                     m_status.unlock();
 
+                    if (control.getEmergency()) {
+                        m_network->subscriberOperations().oneShot(SubscriberOperationStore::Protocol::P25,
+                            SubscriberOperationStore::Kind::EMERG_CALL, srcId, peerId);
+                    }
+
                     // clear any rejected call streams for this TG
                     m_rejectedCallStreams.lock(false);
                     m_rejectedCallStreams[dstId].clear();
@@ -1480,6 +1489,13 @@ bool TagP25Data::processTSDUFrom(uint8_t* buffer, uint32_t peerId, uint8_t duid)
                     }
                 }
                 break;
+            case TSBKO::ISP_EMERG_ALRM_REQ:
+                {
+                    lc::tsbk::ISP_EMERG_ALRM_REQ* isp = static_cast<lc::tsbk::ISP_EMERG_ALRM_REQ*>(tsbk.get());
+                    m_network->subscriberOperations().oneShot(SubscriberOperationStore::Protocol::P25,
+                        SubscriberOperationStore::Kind::EMERG_ALRM, isp->getSrcId(), peerId);
+                }
+                break;
             case TSBKO::IOSP_UU_VCH:
             case TSBKO::IOSP_UU_ANS:
                 {

tests/fne/SubscriberOperationStore_Tests.cpp

@@ -11,6 +11,7 @@
 
 #include <catch2/catch_test_macros.hpp>
 
+#include <string>
 #include <vector>
 
 using namespace network;
@@ -82,3 +83,26 @@ TEST_CASE("FNE subscriber operations return recent history newest first", "[fne]
     CHECK(operations[1U].id == secondId);
     CHECK(store.getLast(0U).empty());
 }
+
+TEST_CASE("FNE subscriber one-shot operations suppress retained duplicates", "[fne][subscriber-operations]")
+{
+    SubscriberOperationStore store;
+    const uint32_t targetId = 0x2345U;
+    const uint64_t operationId = store.oneShot(SubscriberOperationStore::Protocol::NXDN,
+        SubscriberOperationStore::Kind::EMERG_CALL, targetId, 42U);
+
+    REQUIRE(operationId != 0U);
+    CHECK(store.oneShot(SubscriberOperationStore::Protocol::NXDN,
+              SubscriberOperationStore::Kind::EMERG_CALL, targetId, 42U) == 0U);
+    CHECK(store.oneShot(SubscriberOperationStore::Protocol::NXDN,
+              SubscriberOperationStore::Kind::EMERG_CALL, targetId, 43U) == 0U);
+    CHECK(store.oneShot(SubscriberOperationStore::Protocol::DMR,
+              SubscriberOperationStore::Kind::EMERG_CALL, targetId, 42U) != 0U);
+
+    SubscriberOperationStore::Operation operation;
+    REQUIRE(store.get(operationId, operation));
+    CHECK(operation.state == SubscriberOperationStore::State::SUCCEEDED);
+    CHECK(operation.completedAt == operation.createdAt);
+    CHECK(std::string(SubscriberOperationStore::protocolName(operation.protocol)) == "nxdn");
+    CHECK(std::string(SubscriberOperationStore::kindName(operation.kind)) == "emerg-call");
+}