diff --git a/src/common/network/RTPFNEHeader.h b/src/common/network/RTPFNEHeader.h index f8640d11..767be262 100644 --- a/src/common/network/RTPFNEHeader.h +++ b/src/common/network/RTPFNEHeader.h @@ -117,6 +117,7 @@ namespace network SUBSCRIBER_OP_OPERATION = 0x00U, //!< Subscriber Operation Result Query SUBSCRIBER_OP_P25_STATUS = 0x01U, //!< P25 Subscriber Status Query + SUBSCRIBER_OP_START = 0x02U, //!< Start Subscriber Operation ANNC_SUBFUNC_GRP_AFFIL = 0x00U, //!< Announce Group Affiliation ANNC_SUBFUNC_UNIT_REG = 0x01U, //!< Announce Unit Registration diff --git a/src/fne/network/packet/SubscriberOperation.cpp b/src/fne/network/packet/SubscriberOperation.cpp index cfe52033..8ae11de6 100644 --- a/src/fne/network/packet/SubscriberOperation.cpp +++ b/src/fne/network/packet/SubscriberOperation.cpp @@ -9,9 +9,13 @@ * @ingroup fne_network */ #include "fne/Defines.h" +#include "common/dmr/DMRDefines.h" #include "common/Log.h" +#include "common/p25/P25Defines.h" #include "common/json/json.h" #include "network/MetadataNetwork.h" +#include "network/callhandler/TagDMRData.h" +#include "network/callhandler/TagP25Data.h" using namespace network; @@ -46,6 +50,7 @@ void MetadataNetwork::PacketHandler::subscriberOperation(TrafficNetwork* network // prepare the response object and helper for writing the response json::object response = json::object(); const bool notFound = false; + const bool found = true; const NET_SUBFUNC::ENUM subFunction = req->fneHeader.getSubFunction(); const auto writeResponse = [network, mdNetwork, connection, subFunction, streamId](json::object responseObj) { json::value responseValue(responseObj); @@ -89,6 +94,147 @@ void MetadataNetwork::PacketHandler::subscriberOperation(TrafficNetwork* network json::object& request = value.get(); switch (subFunction) { + case NET_SUBFUNC::SUBSCRIBER_OP_START: + { + response["type"].set("operation-start"); + if (!request["protocol"].is() || !request["operation"].is() || + !request["targetId"].is()) { + response["found"].set(notFound); + response["error"].set("invalid operation request"); + break; + } + + const std::string protocolName = request["protocol"].get(); + const std::string operationName = request["operation"].get(); + const uint32_t targetId = request["targetId"].get(); + const uint32_t requestedPeerId = request["peerId"].getDefault(0U); + const uint32_t operationPeerId = requestedPeerId != 0U ? requestedPeerId : network->findPeerUnitReg(targetId); + + if (targetId == 0U) { + response["found"].set(notFound); + response["error"].set("invalid target ID"); + break; + } + + SubscriberOperationStore::Protocol protocol; + SubscriberOperationStore::Kind kind; + uint16_t requestValue = 0U; + if (protocolName == "p25") { + protocol = SubscriberOperationStore::Protocol::P25; + + // determine the kind of P25 operation requested + if (operationName == "radio-check") { + kind = SubscriberOperationStore::Kind::RADIO_CHECK; + } else if (operationName == "status-query") { + kind = SubscriberOperationStore::Kind::STATUS_QUERY; + } else if (operationName == "message-update") { + if (!request["value"].is() || operationPeerId == 0U) { + response["found"].set(notFound); + response["error"].set("message update requires a value and known route or explicit peer ID"); + break; + } + kind = SubscriberOperationStore::Kind::MESSAGE_UPDATE; + requestValue = request["value"].get(); + } else if (operationName == "inhibit") { + kind = SubscriberOperationStore::Kind::INHIBIT; + } else if (operationName == "uninhibit") { + kind = SubscriberOperationStore::Kind::UNINHIBIT; + } else { + response["found"].set(notFound); + response["error"].set("unsupported P25 operation"); + break; + } + } else if (protocolName == "dmr") { + protocol = SubscriberOperationStore::Protocol::DMR; + if (!request["slot"].is()) { + response["found"].set(notFound); + response["error"].set("invalid DMR slot"); + break; + } + + // extract and validate the DMR slot from the request + const uint8_t slot = request["slot"].get(); + if (slot == 0U || slot > 2U) { + response["found"].set(notFound); + response["error"].set("invalid DMR slot"); + break; + } + + // determine the kind of DMR operation requested + if (operationName == "radio-check") { + kind = SubscriberOperationStore::Kind::RADIO_CHECK; + } else if (operationName == "inhibit") { + kind = SubscriberOperationStore::Kind::INHIBIT; + } else if (operationName == "uninhibit") { + kind = SubscriberOperationStore::Kind::UNINHIBIT; + } else { + response["found"].set(notFound); + response["error"].set("unsupported DMR operation"); + break; + } + } else { + response["found"].set(notFound); + response["error"].set("unsupported protocol"); + break; + } + + // start the subscriber operation and obtain its operation ID + const uint64_t operationId = network->subscriberOperations().start( + protocol, kind, targetId, operationPeerId, requestValue); + if (operationId == 0U) { + response["found"].set(notFound); + response["error"].set("an equivalent operation is already pending"); + break; + } + + // send the appropriate protocol-specific message based on the operation kind + if (protocol == SubscriberOperationStore::Protocol::P25) { + switch (kind) { + case SubscriberOperationStore::Kind::RADIO_CHECK: + network->p25TrafficHandler()->write_TSDU_Ext_Func(operationPeerId, + p25::defines::ExtendedFunctions::CHECK, p25::defines::WUID_FNE, targetId); + break; + case SubscriberOperationStore::Kind::STATUS_QUERY: + network->p25TrafficHandler()->write_TSDU_Status_Query(operationPeerId, targetId); + break; + case SubscriberOperationStore::Kind::MESSAGE_UPDATE: + network->p25TrafficHandler()->write_TSDU_Message_Update(operationPeerId, targetId, requestValue); + break; + case SubscriberOperationStore::Kind::INHIBIT: + network->p25TrafficHandler()->write_TSDU_Ext_Func(operationPeerId, + p25::defines::ExtendedFunctions::INHIBIT, p25::defines::WUID_FNE, targetId); + break; + case SubscriberOperationStore::Kind::UNINHIBIT: + network->p25TrafficHandler()->write_TSDU_Ext_Func(operationPeerId, + p25::defines::ExtendedFunctions::UNINHIBIT, p25::defines::WUID_FNE, targetId); + break; + } + } else { + const uint8_t slot = request["slot"].get(); + uint8_t function = dmr::defines::ExtendedFunctions::CHECK; + uint32_t argument = dmr::defines::WUID_ALL; + if (kind == SubscriberOperationStore::Kind::INHIBIT) { + function = dmr::defines::ExtendedFunctions::INHIBIT; + argument = dmr::defines::WUID_STUNI; + } else if (kind == SubscriberOperationStore::Kind::UNINHIBIT) { + function = dmr::defines::ExtendedFunctions::UNINHIBIT; + argument = dmr::defines::WUID_STUNI; + } + + network->dmrTrafficHandler()->write_Ext_Func(operationPeerId, slot, function, argument, targetId); + } + + response["found"].set(found); + response["operationId"].set(operationId); + response["protocol"].set(SubscriberOperationStore::protocolName(protocol)); + response["operation"].set(SubscriberOperationStore::kindName(kind)); + response["state"].set("pending"); + response["targetId"].set(targetId); + response["peerId"].set(operationPeerId); + response["requestValue"].set(requestValue); + } + break; + case NET_SUBFUNC::SUBSCRIBER_OP_OPERATION: { response["type"].set("operation"); diff --git a/tests/fne/FNENetwork_Tests.cpp b/tests/fne/FNENetwork_Tests.cpp index 46b682ea..90e5a20b 100644 --- a/tests/fne/FNENetwork_Tests.cpp +++ b/tests/fne/FNENetwork_Tests.cpp @@ -147,6 +147,94 @@ TEST_CASE("FNE metadata returns subscriber operation and P25 status results", CHECK(statusResponse["solicited"].get()); } +TEST_CASE("FNE metadata starts subscriber operations before transmission", + "[fne][metadata][subscriber-operations]") +{ + constexpr uint32_t CONSOLE_PEER_ID = 1001U; + constexpr uint32_t RADIO_PEER_ID = 2001U; + constexpr uint32_t SUBSCRIBER_ID = 123456U; + const uint8_t dmrSlot = 1U; + + FNEHarness harness; + network::FNEPeerConnection& console = FNETestHooks::addPeer( + harness.traffic, CONSOLE_PEER_ID, network::NET_STAT_RUNNING, true); + console.peerClass(network::PEER_CONN_CLASS_CONSOLE); + FNETestHooks::addPeer(harness.traffic, RADIO_PEER_ID, network::NET_STAT_RUNNING, true); + + std::vector responsePayload; + uint8_t responseFunction = network::NET_FUNC::ILLEGAL; + uint8_t responseSubFunction = network::NET_SUBFUNC::NOP; + FNETestHooks::captureMetadataWrite(harness.metadata, responsePayload, + responseFunction, responseSubFunction); + std::vector peerWrites; + FNETestHooks::capturePeerWrites(harness.traffic, peerWrites); + + json::object request = json::object(); + request["protocol"].set("p25"); + request["operation"].set("radio-check"); + request["targetId"].set(SUBSCRIBER_ID); + request["peerId"].set(RADIO_PEER_ID); + const std::string requestJson = json::value(request).serialize(); + std::vector requestPayload(11U + requestJson.length(), 0U); + ::memcpy(requestPayload.data() + 11U, requestJson.data(), requestJson.length()); + + FNETestHooks::dispatchMetadata(harness.traffic, harness.metadata, + network::NET_FUNC::SUBSCRIBER_OP, network::NET_SUBFUNC::SUBSCRIBER_OP_START, + CONSOLE_PEER_ID, requestPayload); + + REQUIRE(responsePayload.size() > 11U); + CHECK(responseFunction == network::NET_FUNC::SUBSCRIBER_OP); + CHECK(responseSubFunction == network::NET_SUBFUNC::SUBSCRIBER_OP_START); + REQUIRE(peerWrites.size() == 1U); + CHECK(peerWrites.front() == RADIO_PEER_ID); + + json::value responseValue; + REQUIRE(json::parse(responseValue, + std::string(reinterpret_cast(responsePayload.data() + 11U), + responsePayload.size() - 11U)) + .empty()); + REQUIRE(responseValue.is()); + json::object& response = responseValue.get(); + CHECK(response["found"].get()); + CHECK(response["state"].get() == "pending"); + CHECK(response["targetId"].get() == SUBSCRIBER_ID); + CHECK(response["peerId"].get() == RADIO_PEER_ID); + + const uint64_t operationId = response["operationId"].get(); + network::SubscriberOperationStore::Operation operation; + REQUIRE(harness.traffic.subscriberOperations().get(operationId, operation)); + CHECK(operation.kind == network::SubscriberOperationStore::Kind::RADIO_CHECK); + CHECK(operation.state == network::SubscriberOperationStore::State::PENDING); + + request["protocol"].set("dmr"); + request["slot"].set(dmrSlot); + const std::string dmrRequestJson = json::value(request).serialize(); + requestPayload.assign(11U + dmrRequestJson.length(), 0U); + ::memcpy(requestPayload.data() + 11U, dmrRequestJson.data(), dmrRequestJson.length()); + responsePayload.clear(); + peerWrites.clear(); + + FNETestHooks::dispatchMetadata(harness.traffic, harness.metadata, + network::NET_FUNC::SUBSCRIBER_OP, network::NET_SUBFUNC::SUBSCRIBER_OP_START, + CONSOLE_PEER_ID, requestPayload); + + REQUIRE(responsePayload.size() > 11U); + REQUIRE(peerWrites.size() == 1U); + CHECK(peerWrites.front() == RADIO_PEER_ID); + REQUIRE(json::parse(responseValue, + std::string(reinterpret_cast(responsePayload.data() + 11U), + responsePayload.size() - 11U)) + .empty()); + REQUIRE(responseValue.is()); + json::object& dmrResponse = responseValue.get(); + CHECK(dmrResponse["found"].get()); + const uint64_t dmrOperationId = dmrResponse["operationId"].get(); + REQUIRE(harness.traffic.subscriberOperations().get(dmrOperationId, operation)); + CHECK(operation.protocol == network::SubscriberOperationStore::Protocol::DMR); + CHECK(operation.kind == network::SubscriberOperationStore::Kind::RADIO_CHECK); + CHECK(operation.state == network::SubscriberOperationStore::State::PENDING); +} + TEST_CASE("FNE consumes radio-check responses without peer fanout", "[fne][subscriber-operations][routing]") {