add metadata network support for starting P25 and DMR subscriber operations, route protocol messages to the requested or registered peer and add operation start regression tests;

pull/135/head
Bryan Biedenkapp 2 days ago
parent 50cf3b5e7e
commit 2de89d8088

@ -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

@ -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<json::object>();
switch (subFunction) {
case NET_SUBFUNC::SUBSCRIBER_OP_START:
{
response["type"].set<std::string>("operation-start");
if (!request["protocol"].is<std::string>() || !request["operation"].is<std::string>() ||
!request["targetId"].is<uint32_t>()) {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("invalid operation request");
break;
}
const std::string protocolName = request["protocol"].get<std::string>();
const std::string operationName = request["operation"].get<std::string>();
const uint32_t targetId = request["targetId"].get<uint32_t>();
const uint32_t requestedPeerId = request["peerId"].getDefault<uint32_t>(0U);
const uint32_t operationPeerId = requestedPeerId != 0U ? requestedPeerId : network->findPeerUnitReg(targetId);
if (targetId == 0U) {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("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<uint16_t>() || operationPeerId == 0U) {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("message update requires a value and known route or explicit peer ID");
break;
}
kind = SubscriberOperationStore::Kind::MESSAGE_UPDATE;
requestValue = request["value"].get<uint16_t>();
} else if (operationName == "inhibit") {
kind = SubscriberOperationStore::Kind::INHIBIT;
} else if (operationName == "uninhibit") {
kind = SubscriberOperationStore::Kind::UNINHIBIT;
} else {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("unsupported P25 operation");
break;
}
} else if (protocolName == "dmr") {
protocol = SubscriberOperationStore::Protocol::DMR;
if (!request["slot"].is<uint8_t>()) {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("invalid DMR slot");
break;
}
// extract and validate the DMR slot from the request
const uint8_t slot = request["slot"].get<uint8_t>();
if (slot == 0U || slot > 2U) {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("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<bool>(notFound);
response["error"].set<std::string>("unsupported DMR operation");
break;
}
} else {
response["found"].set<bool>(notFound);
response["error"].set<std::string>("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<bool>(notFound);
response["error"].set<std::string>("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>();
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<bool>(found);
response["operationId"].set<uint64_t>(operationId);
response["protocol"].set<std::string>(SubscriberOperationStore::protocolName(protocol));
response["operation"].set<std::string>(SubscriberOperationStore::kindName(kind));
response["state"].set<std::string>("pending");
response["targetId"].set<uint32_t>(targetId);
response["peerId"].set<uint32_t>(operationPeerId);
response["requestValue"].set<uint16_t>(requestValue);
}
break;
case NET_SUBFUNC::SUBSCRIBER_OP_OPERATION:
{
response["type"].set<std::string>("operation");

@ -147,6 +147,94 @@ TEST_CASE("FNE metadata returns subscriber operation and P25 status results",
CHECK(statusResponse["solicited"].get<bool>());
}
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<uint8_t> responsePayload;
uint8_t responseFunction = network::NET_FUNC::ILLEGAL;
uint8_t responseSubFunction = network::NET_SUBFUNC::NOP;
FNETestHooks::captureMetadataWrite(harness.metadata, responsePayload,
responseFunction, responseSubFunction);
std::vector<uint32_t> peerWrites;
FNETestHooks::capturePeerWrites(harness.traffic, peerWrites);
json::object request = json::object();
request["protocol"].set<std::string>("p25");
request["operation"].set<std::string>("radio-check");
request["targetId"].set<uint32_t>(SUBSCRIBER_ID);
request["peerId"].set<uint32_t>(RADIO_PEER_ID);
const std::string requestJson = json::value(request).serialize();
std::vector<uint8_t> 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<const char*>(responsePayload.data() + 11U),
responsePayload.size() - 11U))
.empty());
REQUIRE(responseValue.is<json::object>());
json::object& response = responseValue.get<json::object>();
CHECK(response["found"].get<bool>());
CHECK(response["state"].get<std::string>() == "pending");
CHECK(response["targetId"].get<uint32_t>() == SUBSCRIBER_ID);
CHECK(response["peerId"].get<uint32_t>() == RADIO_PEER_ID);
const uint64_t operationId = response["operationId"].get<uint64_t>();
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<std::string>("dmr");
request["slot"].set<uint8_t>(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<const char*>(responsePayload.data() + 11U),
responsePayload.size() - 11U))
.empty());
REQUIRE(responseValue.is<json::object>());
json::object& dmrResponse = responseValue.get<json::object>();
CHECK(dmrResponse["found"].get<bool>());
const uint64_t dmrOperationId = dmrResponse["operationId"].get<uint64_t>();
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]")
{

Loading…
Cancel
Save

Powered by TurnKey Linux.