implement time-limited FNE subscriber operations for P25 and DMR radio checks, P25 status queries, status updates and message updates, expose operation and status retrieval over REST and metadata networks, add dvmcmd queries, consume acknowledgements without peer fanout and add protocol, routing and store regression tests;
Bryan Biedenkapp
2 days ago
parent
5e14b54b14
commit
080d4d5714
@ -0,0 +1,33 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Common Library
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
#include "common/p25/lc/tsbk/ISP_STS_Q_RSP.h"
|
||||
|
||||
using namespace p25::defines;
|
||||
using namespace p25::lc::tsbk;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Public Class Members
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/* Initializes a new instance of the ISP_STS_Q_RSP class. */
|
||||
|
||||
ISP_STS_Q_RSP::ISP_STS_Q_RSP() :
|
||||
IOSP_STS_UPDT()
|
||||
{
|
||||
m_lco = TSBKO::ISP_STS_Q_RSP;
|
||||
}
|
||||
|
||||
/* Returns a string that represents the current TSBK. */
|
||||
|
||||
std::string ISP_STS_Q_RSP::toString(bool isp)
|
||||
{
|
||||
(void)isp;
|
||||
return "TSBKO, ISP_STS_Q_RSP (Status Query Response)";
|
||||
}
|
||||
@ -0,0 +1,54 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Common Library
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* @file ISP_STS_Q_RSP.h
|
||||
* @ingroup p25_tsbk
|
||||
* @file ISP_STS_Q_RSP.cpp
|
||||
* @ingroup p25_tsbk
|
||||
*/
|
||||
#if !defined(__P25_LC_TSBK__ISP_STS_Q_RSP_H__)
|
||||
#define __P25_LC_TSBK__ISP_STS_Q_RSP_H__
|
||||
|
||||
#include "common/Defines.h"
|
||||
#include "common/p25/lc/tsbk/IOSP_STS_UPDT.h"
|
||||
|
||||
namespace p25
|
||||
{
|
||||
namespace lc
|
||||
{
|
||||
namespace tsbk
|
||||
{
|
||||
// ---------------------------------------------------------------------------
|
||||
// Class Declaration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* @brief Implements STS Q RSP - Status Query Response (ISP).
|
||||
* @ingroup p25_tsbk
|
||||
*/
|
||||
class DVM_COMMON_API ISP_STS_Q_RSP : public IOSP_STS_UPDT {
|
||||
public:
|
||||
/**
|
||||
* @brief Initializes a new instance of the ISP_STS_Q_RSP class.
|
||||
*/
|
||||
ISP_STS_Q_RSP();
|
||||
|
||||
/**
|
||||
* @brief Returns a string that represents the current TSBK.
|
||||
* @param[in] isp Flag indicating ISP formatting; ignored for this fixed-direction TSBK.
|
||||
* @returns std::string String representation of the TSBK.
|
||||
*/
|
||||
std::string toString(bool isp = false) override;
|
||||
};
|
||||
} // namespace tsbk
|
||||
} // namespace lc
|
||||
} // namespace p25
|
||||
|
||||
#endif // __P25_LC_TSBK__ISP_STS_Q_RSP_H__
|
||||
@ -0,0 +1,41 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Common Library
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
#include "common/p25/lc/tsbk/OSP_STS_Q.h"
|
||||
|
||||
using namespace p25::defines;
|
||||
using namespace p25::lc::tsbk;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Public Class Members
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/* Initializes a new instance of the OSP_STS_Q class. */
|
||||
|
||||
OSP_STS_Q::OSP_STS_Q() :
|
||||
IOSP_STS_UPDT()
|
||||
{
|
||||
m_lco = TSBKO::OSP_STS_Q;
|
||||
}
|
||||
|
||||
/* Encodes a status query trunking signalling block. */
|
||||
|
||||
void OSP_STS_Q::encode(uint8_t* data, bool rawTSBK, bool noTrellis)
|
||||
{
|
||||
setStatus(0U);
|
||||
IOSP_STS_UPDT::encode(data, rawTSBK, noTrellis);
|
||||
}
|
||||
|
||||
/* Returns a string that represents the current TSBK. */
|
||||
|
||||
std::string OSP_STS_Q::toString(bool isp)
|
||||
{
|
||||
(void)isp;
|
||||
return "TSBKO, OSP_STS_Q (Status Query)";
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Common Library
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* @file OSP_STS_Q.h
|
||||
* @ingroup p25_tsbk
|
||||
* @file OSP_STS_Q.cpp
|
||||
* @ingroup p25_tsbk
|
||||
*/
|
||||
#if !defined(__P25_LC_TSBK__OSP_STS_Q_H__)
|
||||
#define __P25_LC_TSBK__OSP_STS_Q_H__
|
||||
|
||||
#include "common/Defines.h"
|
||||
#include "common/p25/lc/tsbk/IOSP_STS_UPDT.h"
|
||||
|
||||
namespace p25
|
||||
{
|
||||
namespace lc
|
||||
{
|
||||
namespace tsbk
|
||||
{
|
||||
// ---------------------------------------------------------------------------
|
||||
// Class Declaration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* @brief Implements STS Q - Status Query (OSP).
|
||||
* @ingroup p25_tsbk
|
||||
*/
|
||||
class DVM_COMMON_API OSP_STS_Q : public IOSP_STS_UPDT {
|
||||
public:
|
||||
/**
|
||||
* @brief Initializes a new instance of the OSP_STS_Q class.
|
||||
*/
|
||||
OSP_STS_Q();
|
||||
|
||||
/**
|
||||
* @brief Encodes a status query trunking signalling block.
|
||||
* @param[out] data Buffer to encode a TSBK.
|
||||
* @param[in] rawTSBK Flag indicating whether or not the output buffer is raw.
|
||||
* @param[in] noTrellis Flag indicating whether or not Trellis encoding is suppressed.
|
||||
*/
|
||||
void encode(uint8_t* data, bool rawTSBK = false, bool noTrellis = false) override;
|
||||
|
||||
/**
|
||||
* @brief Returns a string that represents the current TSBK.
|
||||
* @param[in] isp Flag indicating ISP formatting; ignored for this fixed-direction TSBK.
|
||||
* @returns std::string String representation of the TSBK.
|
||||
*/
|
||||
std::string toString(bool isp = false) override;
|
||||
};
|
||||
} // namespace tsbk
|
||||
} // namespace lc
|
||||
} // namespace p25
|
||||
|
||||
#endif // __P25_LC_TSBK__OSP_STS_Q_H__
|
||||
@ -0,0 +1,250 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Converged FNE Software
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
#include "fne/network/SubscriberOperationStore.h"
|
||||
|
||||
using namespace network;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const std::chrono::seconds OPERATION_TIMEOUT(8U);
|
||||
const std::chrono::minutes RESULT_RETENTION(5U);
|
||||
const std::chrono::minutes STATUS_RETENTION(15U);
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Public Class Members
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/* Initializes an empty operation store. */
|
||||
|
||||
SubscriberOperationStore::SubscriberOperationStore() :
|
||||
m_nextId(1U),
|
||||
m_operations(),
|
||||
m_p25Statuses(),
|
||||
m_mutex()
|
||||
{
|
||||
/* stub */
|
||||
}
|
||||
|
||||
/* Starts a uniquely correlatable operation. */
|
||||
|
||||
uint64_t SubscriberOperationStore::start(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);
|
||||
|
||||
// check for existing pending operation with the same protocol, kind, and target ID
|
||||
for (const auto& entry : m_operations) {
|
||||
const Operation& operation = entry.second.value;
|
||||
if (operation.state == State::PENDING && 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::PENDING, targetId, peerId, requestValue, 0U, createdAt, 0U };
|
||||
m_operations.emplace(id, StoredOperation { operation, now + OPERATION_TIMEOUT, now + RESULT_RETENTION });
|
||||
return id;
|
||||
}
|
||||
|
||||
/* Completes the matching active operation. */
|
||||
|
||||
bool SubscriberOperationStore::complete(Protocol protocol, Kind kind, uint32_t targetId, uint32_t peerId,
|
||||
uint16_t resultValue, bool success)
|
||||
{
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
// expire any operations that have timed out or exceeded retention before attempting completion
|
||||
expireLocked(now);
|
||||
|
||||
// find the matching pending operation and complete it
|
||||
for (auto& entry : m_operations) {
|
||||
StoredOperation& stored = entry.second;
|
||||
Operation& operation = stored.value;
|
||||
if (operation.state == State::PENDING && operation.protocol == protocol && operation.kind == kind && operation.targetId == targetId &&
|
||||
(operation.peerId == 0U || operation.peerId == peerId)) {
|
||||
operation.state = success ? State::SUCCEEDED : State::REJECTED;
|
||||
operation.peerId = peerId;
|
||||
operation.resultValue = resultValue;
|
||||
operation.completedAt = unixMilliseconds();
|
||||
stored.retainUntil = now + RESULT_RETENTION;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/* Adds a status report and completes a matching P25 status query. */
|
||||
|
||||
bool SubscriberOperationStore::addP25Status(uint32_t subscriberId, uint32_t peerId, uint16_t value)
|
||||
{
|
||||
const bool solicited = complete(Protocol::P25, Kind::STATUS_QUERY, subscriberId, peerId, value, true);
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
// expire any operations that have timed out or exceeded retention before recording the status
|
||||
expireLocked(now);
|
||||
|
||||
// record the status report with the current timestamp and retention time
|
||||
Status status { subscriberId, peerId, value, solicited, unixMilliseconds() };
|
||||
m_p25Statuses[subscriberId] = StoredStatus { status, now + STATUS_RETENTION };
|
||||
return solicited;
|
||||
}
|
||||
|
||||
/* Retrieves an operation snapshot. */
|
||||
|
||||
bool SubscriberOperationStore::get(uint64_t id, Operation& operation)
|
||||
{
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
// expire any operations that have timed out or exceeded retention before retrieving the snapshot
|
||||
expireLocked(now);
|
||||
|
||||
// look up the operation by its ID
|
||||
const auto it = m_operations.find(id);
|
||||
if (it == m_operations.end()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
operation = it->second.value;
|
||||
return true;
|
||||
}
|
||||
|
||||
/* Retrieves the latest P25 status snapshot. */
|
||||
|
||||
bool SubscriberOperationStore::getP25Status(uint32_t subscriberId, Status& status)
|
||||
{
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
// expire any operations that have timed out or exceeded retention before retrieving the status
|
||||
expireLocked(now);
|
||||
|
||||
// look up the latest status for the given subscriber ID
|
||||
const auto it = m_p25Statuses.find(subscriberId);
|
||||
if (it == m_p25Statuses.end()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
status = it->second.value;
|
||||
return true;
|
||||
}
|
||||
|
||||
/* Expires pending operations and removes retained results. */
|
||||
|
||||
void SubscriberOperationStore::expire()
|
||||
{
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
std::lock_guard<std::mutex> lock(m_mutex);
|
||||
|
||||
// expire any operations that have timed out or exceeded retention before performing general expiration
|
||||
expireLocked(now);
|
||||
}
|
||||
|
||||
/* Returns a stable protocol name. */
|
||||
|
||||
const char* SubscriberOperationStore::protocolName(Protocol protocol)
|
||||
{
|
||||
return protocol == Protocol::DMR ? "dmr" : "p25";
|
||||
}
|
||||
|
||||
/* Returns a stable operation name. */
|
||||
|
||||
const char* SubscriberOperationStore::kindName(Kind kind)
|
||||
{
|
||||
switch (kind) {
|
||||
case Kind::RADIO_CHECK:
|
||||
return "radio-check";
|
||||
case Kind::STATUS_QUERY:
|
||||
return "status-query";
|
||||
case Kind::MESSAGE_UPDATE:
|
||||
return "message-update";
|
||||
case Kind::INHIBIT:
|
||||
return "inhibit";
|
||||
case Kind::UNINHIBIT:
|
||||
return "uninhibit";
|
||||
}
|
||||
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
/* Returns a stable state name. */
|
||||
|
||||
const char* SubscriberOperationStore::stateName(State state)
|
||||
{
|
||||
switch (state) {
|
||||
case State::PENDING:
|
||||
return "pending";
|
||||
case State::SUCCEEDED:
|
||||
return "succeeded";
|
||||
case State::REJECTED:
|
||||
return "rejected";
|
||||
case State::TIMED_OUT:
|
||||
return "timed-out";
|
||||
}
|
||||
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Private Class Members
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/* Gets the current Unix timestamp. */
|
||||
|
||||
uint64_t SubscriberOperationStore::unixMilliseconds()
|
||||
{
|
||||
return static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count());
|
||||
}
|
||||
|
||||
/* Expires records while the store mutex is held. */
|
||||
|
||||
void SubscriberOperationStore::expireLocked(const std::chrono::steady_clock::time_point& now)
|
||||
{
|
||||
// iterate through all stored operations and expire those that have timed out or exceeded retention
|
||||
for (auto it = m_operations.begin(); it != m_operations.end();) {
|
||||
StoredOperation& stored = it->second;
|
||||
|
||||
// check if the operation has timed out and update its state accordingly
|
||||
if (stored.value.state == State::PENDING && now >= stored.deadline) {
|
||||
stored.value.state = State::TIMED_OUT;
|
||||
stored.value.completedAt = unixMilliseconds();
|
||||
stored.retainUntil = now + RESULT_RETENTION;
|
||||
}
|
||||
|
||||
// remove operations that have completed and exceeded their retention period
|
||||
if (stored.value.state != State::PENDING && now >= stored.retainUntil) {
|
||||
it = m_operations.erase(it);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
|
||||
// iterate through all stored P25 statuses and remove those that have exceeded their retention period
|
||||
for (auto it = m_p25Statuses.begin(); it != m_p25Statuses.end();) {
|
||||
if (now >= it->second.retainUntil) {
|
||||
it = m_p25Statuses.erase(it);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,213 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Converged FNE Software
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
/**
|
||||
* @file SubscriberOperationStore.h
|
||||
* @ingroup fne_network
|
||||
* @file SubscriberOperationStore.cpp
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
#if !defined(__SUBSCRIBER_OPERATION_STORE_H__)
|
||||
#define __SUBSCRIBER_OPERATION_STORE_H__
|
||||
|
||||
#include "fne/Defines.h"
|
||||
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <mutex>
|
||||
#include <unordered_map>
|
||||
|
||||
namespace network
|
||||
{
|
||||
// ---------------------------------------------------------------------------
|
||||
// Class Declaration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* @brief Stores short-lived FNE subscriber-management operations and status observations.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
class HOST_SW_API SubscriberOperationStore {
|
||||
public:
|
||||
/**
|
||||
* @brief Supported air-interface protocols.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
enum class Protocol : uint8_t {
|
||||
DMR = 0U, //!< Digital Mobile Radio.
|
||||
P25 = 1U //!< Project 25.
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief Supported operation kinds.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
enum class Kind : uint8_t {
|
||||
RADIO_CHECK = 0U, //!< Subscriber presence check.
|
||||
STATUS_QUERY = 1U, //!< Subscriber status query.
|
||||
MESSAGE_UPDATE = 2U,//!< Subscriber coded-message update.
|
||||
INHIBIT = 3U, //!< Subscriber inhibit request.
|
||||
UNINHIBIT = 4U //!< Subscriber uninhibit request.
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief Operation lifecycle states.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
enum class State : uint8_t {
|
||||
PENDING = 0U, //!< Request has been transmitted and awaits a response.
|
||||
SUCCEEDED = 1U, //!< A matching successful response was received.
|
||||
REJECTED = 2U, //!< A matching negative response was received.
|
||||
TIMED_OUT = 3U //!< No matching response arrived before the deadline.
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief Immutable snapshot of one operation.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
struct Operation {
|
||||
uint64_t id; //!< FNE-local operation identifier.
|
||||
Protocol protocol; //!< Air-interface protocol.
|
||||
Kind kind; //!< Requested operation.
|
||||
State state; //!< Current lifecycle state.
|
||||
uint32_t targetId; //!< Target subscriber ID.
|
||||
uint32_t peerId; //!< Requested or responding peer ID.
|
||||
uint16_t requestValue; //!< Optional requested status/message value.
|
||||
uint16_t resultValue; //!< Optional returned status value.
|
||||
uint64_t createdAt; //!< Creation time in Unix milliseconds.
|
||||
uint64_t completedAt; //!< Completion time in Unix milliseconds, or zero.
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief Latest status observed for a P25 subscriber.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
struct Status {
|
||||
uint32_t subscriberId; //!< Reporting subscriber ID.
|
||||
uint32_t peerId; //!< Peer that delivered the report.
|
||||
uint16_t value; //!< Status value.
|
||||
bool solicited; //!< True when matched to an active query.
|
||||
uint64_t observedAt; //!< Observation time in Unix milliseconds.
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief Initializes an empty operation store.
|
||||
*/
|
||||
SubscriberOperationStore();
|
||||
|
||||
/**
|
||||
* @brief Starts a uniquely correlatable 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 start(Protocol protocol, Kind kind, uint32_t targetId, uint32_t peerId, uint16_t requestValue = 0U);
|
||||
|
||||
/**
|
||||
* @brief Completes the matching active operation.
|
||||
* @param[in] protocol Air-interface protocol.
|
||||
* @param[in] kind Operation kind.
|
||||
* @param[in] targetId Target subscriber ID.
|
||||
* @param[in] peerId Responding peer ID.
|
||||
* @param[in] resultValue Optional returned value.
|
||||
* @param[in] success True for successful completion.
|
||||
* @returns bool True when an active operation was completed.
|
||||
*/
|
||||
bool complete(Protocol protocol, Kind kind, uint32_t targetId, uint32_t peerId,
|
||||
uint16_t resultValue = 0U, bool success = true);
|
||||
|
||||
/**
|
||||
* @brief Adds a status report and completes a matching P25 status query.
|
||||
* @param[in] subscriberId Reporting subscriber ID.
|
||||
* @param[in] peerId Peer that delivered the report.
|
||||
* @param[in] value Status value.
|
||||
* @returns bool True when the report completed a pending query.
|
||||
*/
|
||||
bool addP25Status(uint32_t subscriberId, uint32_t peerId, uint16_t value);
|
||||
|
||||
/**
|
||||
* @brief Retrieves an operation snapshot.
|
||||
* @param[in] id Operation identifier.
|
||||
* @param[out] operation Operation snapshot.
|
||||
* @returns bool True if the operation exists, otherwise false.
|
||||
*/
|
||||
bool get(uint64_t id, Operation& operation);
|
||||
/**
|
||||
* @brief Retrieves the latest P25 status snapshot.
|
||||
* @param[in] subscriberId Subscriber ID.
|
||||
* @param[out] status Status snapshot.
|
||||
* @returns bool True if an unexpired status exists, otherwise false.
|
||||
*/
|
||||
bool getP25Status(uint32_t subscriberId, Status& status);
|
||||
/**
|
||||
* @brief Expires pending operations and removes retained results.
|
||||
*/
|
||||
void expire();
|
||||
|
||||
/**
|
||||
* @brief Returns a stable protocol name.
|
||||
* @param[in] protocol Protocol value.
|
||||
* @returns const char* Protocol name with static storage duration.
|
||||
*/
|
||||
static const char* protocolName(Protocol protocol);
|
||||
/**
|
||||
* @brief Returns a stable operation name.
|
||||
* @param[in] kind Operation kind.
|
||||
* @returns const char* Operation name with static storage duration.
|
||||
*/
|
||||
static const char* kindName(Kind kind);
|
||||
/**
|
||||
* @brief Returns a stable state name.
|
||||
* @param[in] state Operation state.
|
||||
* @returns const char* State name with static storage duration.
|
||||
*/
|
||||
static const char* stateName(State state);
|
||||
|
||||
private:
|
||||
/**
|
||||
* @brief Internal operation record with monotonic expiry deadlines.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
struct StoredOperation {
|
||||
Operation value; //!< Externally visible operation data.
|
||||
std::chrono::steady_clock::time_point deadline; //!< Monotonic response deadline.
|
||||
std::chrono::steady_clock::time_point retainUntil; //!< Monotonic result retention deadline.
|
||||
};
|
||||
|
||||
/**
|
||||
* @brief Internal subscriber status record with a monotonic expiry deadline.
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
struct StoredStatus {
|
||||
Status value; //!< Externally visible status data.
|
||||
std::chrono::steady_clock::time_point retainUntil; //!< Monotonic status retention deadline.
|
||||
};
|
||||
|
||||
uint64_t m_nextId; //!< Next candidate operation identifier.
|
||||
std::unordered_map<uint64_t, StoredOperation> m_operations; //!< Operations indexed by identifier.
|
||||
std::unordered_map<uint32_t, StoredStatus> m_p25Statuses; //!< Latest P25 status indexed by subscriber ID.
|
||||
std::mutex m_mutex; //!< Protects all store state.
|
||||
|
||||
/**
|
||||
* @brief Gets the current Unix timestamp.
|
||||
* @returns uint64_t Current Unix time in milliseconds.
|
||||
*/
|
||||
static uint64_t unixMilliseconds();
|
||||
/**
|
||||
* @brief Expires records while the store mutex is held.
|
||||
* @param[in] now Current monotonic time.
|
||||
*/
|
||||
void expireLocked(const std::chrono::steady_clock::time_point& now);
|
||||
};
|
||||
} // namespace network
|
||||
|
||||
#endif // __SUBSCRIBER_OPERATION_STORE_H__
|
||||
@ -0,0 +1,152 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Converged FNE Software
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*/
|
||||
/**
|
||||
* @file SubscriberOperation.cpp
|
||||
* @ingroup fne_network
|
||||
*/
|
||||
#include "fne/Defines.h"
|
||||
#include "common/Log.h"
|
||||
#include "common/json/json.h"
|
||||
#include "network/MetadataNetwork.h"
|
||||
|
||||
using namespace network;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const uint32_t SUBSCRIBER_OPERATION_HEADER_LENGTH = 11U;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Public Class Members
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/* Handles NET_FUNC::SUBSCRIBER_OP packets. */
|
||||
|
||||
void MetadataNetwork::PacketHandler::subscriberOperation(TrafficNetwork* network,
|
||||
MetadataNetwork* mdNetwork, NetPacketRequest* req, uint32_t peerId, uint32_t ssrc,
|
||||
uint32_t streamId)
|
||||
{
|
||||
(void)ssrc;
|
||||
|
||||
TrafficNetwork::PeerConnectionPtr connection = network->findPeer(peerId);
|
||||
const std::string sourceAddress = udp::Socket::address(req->address);
|
||||
|
||||
// validate the peer connection and source address before processing the subscriber operation request
|
||||
if (connection == nullptr || !connection->connected() || connection->peerClass() != PEER_CONN_CLASS_CONSOLE ||
|
||||
(!connection->address().empty() && connection->address() != sourceAddress)) {
|
||||
network->writePeerNAK(peerId, network->createStreamId(), "SUBSCRIBER_OP", NET_CONN_NAK_FNE_UNAUTHORIZED);
|
||||
return;
|
||||
}
|
||||
|
||||
// prepare the response object and helper for writing the response
|
||||
json::object response = json::object();
|
||||
const bool notFound = false;
|
||||
const NET_SUBFUNC::ENUM subFunction = req->fneHeader.getSubFunction();
|
||||
const auto writeResponse = [network, mdNetwork, connection, subFunction, streamId](json::object responseObj) {
|
||||
json::value responseValue(responseObj);
|
||||
const std::string responsePayload = responseValue.serialize();
|
||||
const uint32_t responseLength = static_cast<uint32_t>(responsePayload.length());
|
||||
|
||||
// check if the response payload fits within the allowed data packet length
|
||||
if (responseLength + SUBSCRIBER_OPERATION_HEADER_LENGTH > DATA_PACKET_LENGTH) {
|
||||
LogError(LOG_MASTER, "PEER %u subscriber operation response too large, len = %u",
|
||||
connection->id(), responseLength);
|
||||
return false;
|
||||
}
|
||||
|
||||
uint8_t buffer[DATA_PACKET_LENGTH];
|
||||
::memset(buffer, 0x00U, DATA_PACKET_LENGTH);
|
||||
::memcpy(buffer + SUBSCRIBER_OPERATION_HEADER_LENGTH, responsePayload.data(), responseLength);
|
||||
|
||||
return mdNetwork->writePeerMetadata(connection.get(), network->m_peerId, { NET_FUNC::SUBSCRIBER_OP, subFunction }, buffer,
|
||||
responseLength + SUBSCRIBER_OPERATION_HEADER_LENGTH, RTP_END_OF_CALL_SEQ, streamId);
|
||||
};
|
||||
|
||||
// validate the request length against the expected header and data packet limits
|
||||
if (req->length <= static_cast<int>(SUBSCRIBER_OPERATION_HEADER_LENGTH) || req->length > static_cast<int>(DATA_PACKET_LENGTH)) {
|
||||
response["found"].set<bool>(notFound);
|
||||
response["error"].set<std::string>("invalid request");
|
||||
writeResponse(response);
|
||||
return;
|
||||
}
|
||||
|
||||
// calculate the payload length by subtracting the header length from the total request length
|
||||
const size_t payloadLength = static_cast<size_t>(req->length) - SUBSCRIBER_OPERATION_HEADER_LENGTH;
|
||||
const std::string payload(reinterpret_cast<const char*>(req->buffer + SUBSCRIBER_OPERATION_HEADER_LENGTH), payloadLength);
|
||||
json::value value;
|
||||
const std::string parseError = json::parse(value, payload);
|
||||
if (!parseError.empty() || !value.is<json::object>()) {
|
||||
response["found"].set<bool>(notFound);
|
||||
response["error"].set<std::string>("invalid request");
|
||||
writeResponse(response);
|
||||
return;
|
||||
}
|
||||
|
||||
json::object& request = value.get<json::object>();
|
||||
switch (subFunction) {
|
||||
case NET_SUBFUNC::SUBSCRIBER_OP_OPERATION:
|
||||
{
|
||||
response["type"].set<std::string>("operation");
|
||||
if (!request["operationId"].is<uint64_t>()) {
|
||||
response["found"].set<bool>(notFound);
|
||||
response["error"].set<std::string>("invalid operation ID");
|
||||
break;
|
||||
}
|
||||
|
||||
const uint64_t operationId = request["operationId"].get<uint64_t>();
|
||||
SubscriberOperationStore::Operation operation;
|
||||
const bool found = operationId != 0U && network->subscriberOperations().get(operationId, operation);
|
||||
response["found"].set<bool>(found);
|
||||
response["operationId"].set<uint64_t>(operationId);
|
||||
if (found) {
|
||||
response["protocol"].set<std::string>(SubscriberOperationStore::protocolName(operation.protocol));
|
||||
response["operation"].set<std::string>(SubscriberOperationStore::kindName(operation.kind));
|
||||
response["state"].set<std::string>(SubscriberOperationStore::stateName(operation.state));
|
||||
response["targetId"].set<uint32_t>(operation.targetId);
|
||||
response["peerId"].set<uint32_t>(operation.peerId);
|
||||
response["requestValue"].set<uint16_t>(operation.requestValue);
|
||||
response["resultValue"].set<uint16_t>(operation.resultValue);
|
||||
response["createdAt"].set<uint64_t>(operation.createdAt);
|
||||
response["completedAt"].set<uint64_t>(operation.completedAt);
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case NET_SUBFUNC::SUBSCRIBER_OP_P25_STATUS:
|
||||
{
|
||||
response["type"].set<std::string>("p25-status");
|
||||
if (!request["subscriberId"].is<uint32_t>()) {
|
||||
response["found"].set<bool>(notFound);
|
||||
response["error"].set<std::string>("invalid subscriber ID");
|
||||
break;
|
||||
}
|
||||
|
||||
const uint32_t subscriberId = request["subscriberId"].get<uint32_t>();
|
||||
SubscriberOperationStore::Status status;
|
||||
const bool found = subscriberId != 0U &&
|
||||
network->subscriberOperations().getP25Status(subscriberId, status);
|
||||
response["found"].set<bool>(found);
|
||||
response["protocol"].set<std::string>("p25");
|
||||
response["subscriberId"].set<uint32_t>(subscriberId);
|
||||
if (found) {
|
||||
response["peerId"].set<uint32_t>(status.peerId);
|
||||
response["status"].set<uint16_t>(status.value);
|
||||
response["solicited"].set<bool>(status.solicited);
|
||||
response["observedAt"].set<uint64_t>(status.observedAt);
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
response["found"].set<bool>(notFound);
|
||||
response["error"].set<std::string>("unsupported query");
|
||||
break;
|
||||
}
|
||||
|
||||
writeResponse(response);
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
/*
|
||||
* Digital Voice Modem - Test Suite
|
||||
* GPLv2 Open Source. Use is subject to license terms.
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* Copyright (C) 2026 Bryan Biedenkapp, N2PLL
|
||||
*
|
||||
*/
|
||||
#include "fne/network/SubscriberOperationStore.h"
|
||||
|
||||
#include <catch2/catch_test_macros.hpp>
|
||||
|
||||
using namespace network;
|
||||
|
||||
TEST_CASE("FNE subscriber operations correlate protocol results", "[fne][subscriber-operations]")
|
||||
{
|
||||
SubscriberOperationStore store;
|
||||
const uint32_t targetId = 0x123456U;
|
||||
|
||||
uint64_t operationId = store.start(SubscriberOperationStore::Protocol::P25,
|
||||
SubscriberOperationStore::Kind::STATUS_QUERY, targetId, 42U);
|
||||
REQUIRE(operationId != 0U);
|
||||
CHECK(store.start(SubscriberOperationStore::Protocol::P25,
|
||||
SubscriberOperationStore::Kind::STATUS_QUERY, targetId, 42U) == 0U);
|
||||
CHECK_FALSE(store.addP25Status(targetId, 43U, 0x1234U));
|
||||
CHECK(store.addP25Status(targetId, 42U, 0x5678U));
|
||||
|
||||
SubscriberOperationStore::Operation operation;
|
||||
REQUIRE(store.get(operationId, operation));
|
||||
CHECK(operation.state == SubscriberOperationStore::State::SUCCEEDED);
|
||||
CHECK(operation.resultValue == 0x5678U);
|
||||
|
||||
SubscriberOperationStore::Status status;
|
||||
REQUIRE(store.getP25Status(targetId, status));
|
||||
CHECK(status.value == 0x5678U);
|
||||
CHECK(status.solicited);
|
||||
}
|
||||
|
||||
TEST_CASE("FNE subscriber operations isolate protocols and kinds", "[fne][subscriber-operations]")
|
||||
{
|
||||
SubscriberOperationStore store;
|
||||
const uint32_t targetId = 0x654321U;
|
||||
const uint64_t dmrId = store.start(SubscriberOperationStore::Protocol::DMR,
|
||||
SubscriberOperationStore::Kind::RADIO_CHECK, targetId, 0U);
|
||||
const uint64_t p25Id = store.start(SubscriberOperationStore::Protocol::P25,
|
||||
SubscriberOperationStore::Kind::RADIO_CHECK, targetId, 0U);
|
||||
|
||||
REQUIRE(dmrId != 0U);
|
||||
REQUIRE(p25Id != 0U);
|
||||
REQUIRE(store.complete(SubscriberOperationStore::Protocol::DMR,
|
||||
SubscriberOperationStore::Kind::RADIO_CHECK, targetId, 12U));
|
||||
|
||||
SubscriberOperationStore::Operation dmrOperation;
|
||||
SubscriberOperationStore::Operation p25Operation;
|
||||
REQUIRE(store.get(dmrId, dmrOperation));
|
||||
REQUIRE(store.get(p25Id, p25Operation));
|
||||
CHECK(dmrOperation.state == SubscriberOperationStore::State::SUCCEEDED);
|
||||
CHECK(dmrOperation.peerId == 12U);
|
||||
CHECK(p25Operation.state == SubscriberOperationStore::State::PENDING);
|
||||
}
|
||||
Loading…
Reference in new issue