diff --git a/AGENTS.md b/AGENTS.md index c67a392ac..c04beb662 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -470,10 +470,17 @@ selected at connect time by a `family` string through `makeBackend()`: Step 3 is in progress: the normative v1 envelope contract, bounded codec, observe-only local handshake/capability service, and a QtWidgets-free -`aetherd` skeleton have landed. Typed model resources, subscriptions, -authenticated non-TX control, and the desktop adapter have not; UI code still -consumes models directly, and that remains correct. No protocol TX method is -advertised before the step-4 arbiter exists. +`aetherd` skeleton have landed. The typed observe-only `server`, +`radioSession`, `slice`, and `panadapter` resources now publish through +`RadioResourceAdapter`; `resource.get` plus atomic snapshot/event +`resource.subscribe`/`resource.unsubscribe`, per-resource revisions, bounded +coalescing/session resync, and an independent local-socket hard disconnect cap +are live over the current-user local transport. +Meters, read-only transmit state, authenticated non-TX control, and the desktop +adapter have not landed; UI code still consumes models directly, and that +remains correct. New resource fields belong in the adapter and the versioned +catalogue, never in a transport or via QObject reflection. No protocol TX +method is advertised before the step-4 arbiter exists. **Backends that demodulate in-process double-feed the sink if you let them.** `IRadioBackend::audioFrameReady` has two possible routes to diff --git a/CMakeLists.txt b/CMakeLists.txt index 04dec1648..aa5bf6f0d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -683,7 +683,10 @@ set(AETHER_SETTINGS_SOURCES set(CORE_SOURCES src/core/control/ControlProtocolCodec.cpp # aetherd v1 bounded JSON envelopes + src/core/control/ControlResourceStore.cpp # typed snapshots and revisions + src/core/control/ControlSession.cpp # subscriptions and bounded event queues src/core/control/ControlService.cpp # aetherd v1 observe-only dispatch + src/core/control/RadioResourceAdapter.cpp # normalized model resources src/core/control/LocalControlServer.cpp # current-user local transport src/core/backends/MemoryWireCodec.cpp # memory kv-set decode, shared by Flex + local bank src/core/backends/flex/FlexBackend.cpp # aetherd RFC step 2.2 (§5.5) diff --git a/docs/aetherd-control-protocol-v1-design.md b/docs/aetherd-control-protocol-v1-design.md index 555ddba2c..c6ffe2e3e 100644 --- a/docs/aetherd-control-protocol-v1-design.md +++ b/docs/aetherd-control-protocol-v1-design.md @@ -186,6 +186,9 @@ schema test before it is exported. Credentials, backend pointers, raw vendor messages, filesystem paths, unbounded log text, and other clients' private state are never resources. +The schemas implemented by the first observe-only slice are fixed in +[`aetherd-control-resource-v1-catalogue.md`](aetherd-control-resource-v1-catalogue.md). + `resource.subscribe` accepts explicit resource selectors and returns an atomic baseline: @@ -200,11 +203,13 @@ baseline: } ``` -The service registers the subscription, captures the snapshot, and queues all -changes after the returned `sequence` as one main-thread operation. Therefore -there is no snapshot/event gap. Reconnect creates a new protocol session; -clients resubscribe and replace their cache from a fresh snapshot. V1 does not -promise event replay across connections. +The service returns the last event sequence already drained to the transport, +registers the subscription, and captures the snapshot as one main-thread +operation. Events still pending for existing subscriptions retain greater +sequences, and newly generated changes advance beyond them. Therefore there is +no snapshot/event gap or duplicate baseline sequence. Reconnect creates a new +protocol session; clients resubscribe and replace their cache from a fresh +snapshot. V1 does not promise event replay across connections. High-rate spectrum, waterfall and audio payloads are not embedded in these control JSON events. Their later stream contract must use bounded binary diff --git a/docs/aetherd-control-resource-v1-catalogue.md b/docs/aetherd-control-resource-v1-catalogue.md new file mode 100644 index 000000000..67c4cf7aa --- /dev/null +++ b/docs/aetherd-control-resource-v1-catalogue.md @@ -0,0 +1,169 @@ +# AetherD control protocol v1 — observe-only resource catalogue + +This catalogue fixes the schema implemented by the first read-only Stage 3 +slice of RFC #3849. It supplements +[`aetherd-control-protocol-v1-design.md`](aetherd-control-protocol-v1-design.md); +the envelope, limits, errors, authentication, and TX rules in that document +remain normative. + +Only the four resource types below exist in this slice. `meter` and +`transmitState` remain unimplemented. No method in this catalogue mutates a +model or can reach a radio backend intent. + +## Resource identities and selectors + +An exact identity has one of these shapes: + +```json +{"type":"server"} +{"type":"radioSession","id":"radio-1"} +{"type":"slice","radioSession":"radio-1","id":"0"} +{"type":"panadapter","radioSession":"radio-1","id":"0x40000000"} +``` + +`resource.get` requires an exact identity. `resource.subscribe` also accepts +an omitted `id` as an all-current-and-future selector for `radioSession`, +`slice`, or `panadapter`; `slice` and `panadapter` still require +`radioSession`. Unknown fields and unsupported resource types are rejected. + +## Methods + +All methods require the negotiated session ID. The initial current-user local +endpoint grants `observe` to every negotiated session; because no other session +type exists yet, this slice has no separate per-request grant branch. Explicit +per-session grant mapping and checks arrive with authentication before another +grant or remote session is exposed. + +### `resource.get` + +Parameters: + +```json +{"resource":{"type":"slice","radioSession":"radio-1","id":"0"}} +``` + +Result: + +```json +{ + "resource":{"type":"slice","radioSession":"radio-1","id":"0"}, + "revision":3, + "value":{} +} +``` + +The complete typed value occupies `value`. A missing exact identity returns +`resource.not_found`. + +### `resource.subscribe` + +Parameters contain 1–64 selectors: + +```json +{"resources":[{"type":"slice","radioSession":"radio-1"}]} +``` + +The result contains a session-local subscription ID, the last session event +sequence already drained to the transport, and the complete baseline matching +those selectors. Registration and snapshot capture execute as one main-thread +operation. Events still pending for existing subscriptions retain sequences +greater than the returned boundary, and newly generated events advance beyond +them, so an event delivered after the baseline cannot leave a snapshot/event +gap or reuse the baseline sequence. + +### `resource.unsubscribe` + +Parameters are `{"subscription":"sub-1"}`. Success returns the same ID and +`"removed":true`. An unknown ID returns `resource.not_found`. + +## Events, revisions, and resync + +`resource.changed` carries the complete new value. `resource.removed` carries +the identity and its next revision but no value. Revisions come from one +store-wide monotonic counter. They are therefore monotonic per identity and +survive removal/recreation. A revision is consumed only when a canonical value +changes or a live identity is removed, but an identity's revisions need not be +consecutive or begin at one. + +Event `sequence` is monotonic within one protocol session. Pending events for +the same resource coalesce to the newest sequence, revision, and complete value. +Sequences may therefore have gaps; they never move backward. + +If a session's bounded event queue cannot retain its subscribed state, the +service clears that session's subscriptions and emits: + +```json +{ + "v":1, + "sessionId":"...", + "event":"resource.resyncRequired", + "sequence":42, + "subscriptionsInvalidated":true +} +``` + +The client must call `resource.subscribe` again and replace its cache from the +fresh baseline. The current-user local transport also enforces a hard +socket-output cap. That cap is a separate *check* — the session's pending queue +against the operating-system socket buffer — but not a separate *budget*: both +are the same `maxQueuedOutputBytes` figure advertised in the handshake, so a +client should budget that figure once, not twice. A client whose socket buffer +is already at the cap can be disconnected before a queued resync notice is +written; after reconnecting it must establish a new session and baseline. + +## Resource values + +### `server` + +- `name`: server product name. +- `buildVersion`: AetherSDR build version. +- `protocolVersions`: supported protocol versions. +- `health`: bounded service health token. +- `localTransport`: `idle`, `listening`, or `stopped`. `idle` and `stopped` + describe in-process lifecycle state before or after socket availability; a + protocol client can query this resource only while the value is `listening`. + +No endpoint path, process environment, hostname, or filesystem value is +exported. + +### `radioSession` + +- `id`, `connected`, `family`. +- `identity`: `name`, `model`, `serial`, `version`, `manufacturer`. +- `capabilities`: + - `maxSlices`, `maxPanadapters`, `sampleRatesHz`; + - `tuningRangeHz` with `minimum` and `maximum`; + - `declaredBands`, each with `name`, `lowHz`, and `highHz`; + - `canTransmit`, `maximumTransmitWatts`, `hasTuner`, `hasAmplifier`; + - `extensions`, containing namespace names only, never extension payloads. + +`canTransmit` is observation only. It does not advertise a protocol TX method +or grant and cannot key a radio. + +### `slice` + +- `id`, `letter`, `panadapterId`, `owned`. +- `frequencyHz`, `mode`, `filter.lowHz`, `filter.highHz`. +- `active`, `txSlice`, `locked`. +- `audio.gain`, `audio.pan`, `audio.muted`. +- `receive.antenna`, `receive.rfGain`. +- `receive.agc.mode`, `receive.agc.threshold`, `receive.agc.offLevel`. +- `receive.squelch.enabled`, `receive.squelch.level`. + +Values come from `SliceModel`; radio/backend status remains authoritative. + +### `panadapter` + +- `id`. +- `centerHz`, `centerKnown`, `bandwidthHz`. +- `dbmRange.minimum`, `dbmRange.maximum`. +- `bandwidthLimitsHz.minimum`, `bandwidthLimitsHz.maximum`; zero means the + backend has not reported a limit. +- `receive.antenna`, `receive.rfGain`. +- `displayCadence.fps`, `displayCadence.averageFrames`. +- `displayCadence.weightedAverage`, `weightedAverageKnown`. +- `displayCadence.waterfallRate`; `-1` means the backend has not reported a + value, otherwise this is the normalized 1–100 rate, not milliseconds. + +FFT bins, waterfall rows, audio, and other high-rate data never enter these +JSON resources; they belong to the later bounded binary data plane. diff --git a/src/aetherd/main.cpp b/src/aetherd/main.cpp index 980b6352b..b26266f24 100644 --- a/src/aetherd/main.cpp +++ b/src/aetherd/main.cpp @@ -1,4 +1,6 @@ #include "core/control/LocalControlServer.h" +#include "core/control/RadioResourceAdapter.h" +#include "models/RadioSession.h" #include #include @@ -22,7 +24,12 @@ int main(int argc, char* argv[]) parser.addOption(socketOption); parser.process(app); + AetherSDR::RadioSession radioSession; + radioSession.setSessionId(1); AetherSDR::control::LocalControlServer server; + [[maybe_unused]] AetherSDR::control::RadioResourceAdapter resources( + &radioSession.radioModel(), &server.resourceStore(), + QStringLiteral("radio-1")); if (!server.listen(parser.value(socketOption))) { QTextStream(stderr) << "aetherd: cannot listen on local socket '" << parser.value(socketOption) << "'\n"; diff --git a/src/core/control/ControlResourceStore.cpp b/src/core/control/ControlResourceStore.cpp new file mode 100644 index 000000000..942828dba --- /dev/null +++ b/src/core/control/ControlResourceStore.cpp @@ -0,0 +1,93 @@ +#include "ControlResourceStore.h" + +namespace AetherSDR::control { + +QString ResourceAddress::key() const +{ + return type + QChar(0x1f) + radioSession + QChar(0x1f) + id; +} + +QJsonObject ResourceAddress::toJson() const +{ + QJsonObject object{{QStringLiteral("type"), type}}; + if (!radioSession.isEmpty()) { + object.insert(QStringLiteral("radioSession"), radioSession); + } + if (!id.isEmpty()) { + object.insert(QStringLiteral("id"), id); + } + return object; +} + +bool ResourceSelector::matches(const ResourceAddress& address) const +{ + return type == address.type + && (radioSession.isEmpty() || radioSession == address.radioSession) + && (id.isEmpty() || id == address.id); +} + +QJsonObject ResourceSnapshot::toJson() const +{ + return {{QStringLiteral("resource"), resource.toJson()}, + {QStringLiteral("revision"), static_cast(revision)}, + {QStringLiteral("value"), value}}; +} + +ControlResourceStore::ControlResourceStore(QObject* parent) + : QObject(parent) +{ +} + +bool ControlResourceStore::upsert( + const ResourceAddress& address, const QJsonObject& value) +{ + const QString resourceKey = address.key(); + const auto current = m_resources.constFind(resourceKey); + if (current != m_resources.constEnd() && current->value == value) { + return false; + } + + const quint64 revision = ++m_lastRevision; + const ResourceSnapshot next{address, revision, value}; + m_resources.insert(resourceKey, next); + emit resourceChanged(next); + return true; +} + +bool ControlResourceStore::remove(const ResourceAddress& address) +{ + const QString resourceKey = address.key(); + if (m_resources.remove(resourceKey) == 0) { + return false; + } + const quint64 revision = ++m_lastRevision; + emit resourceRemoved(address, revision); + return true; +} + +std::optional ControlResourceStore::get( + const ResourceAddress& address) const +{ + const auto found = m_resources.constFind(address.key()); + if (found == m_resources.constEnd()) { + return std::nullopt; + } + return *found; +} + +QList ControlResourceStore::snapshot( + const QList& selectors) const +{ + QList result; + for (auto it = m_resources.constBegin(); it != m_resources.constEnd(); ++it) { + for (const ResourceSelector& selector : selectors) { + if (selector.matches(it->resource)) { + result.append(*it); + break; + } + } + } + return result; +} + +} // namespace AetherSDR::control diff --git a/src/core/control/ControlResourceStore.h b/src/core/control/ControlResourceStore.h new file mode 100644 index 000000000..beb562acf --- /dev/null +++ b/src/core/control/ControlResourceStore.h @@ -0,0 +1,70 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include + +namespace AetherSDR::control { + +struct ResourceAddress { + QString type; + QString radioSession; + QString id; + + [[nodiscard]] QString key() const; + [[nodiscard]] QJsonObject toJson() const; + bool operator==(const ResourceAddress&) const = default; +}; + +struct ResourceSelector { + QString type; + QString radioSession; + QString id; + + [[nodiscard]] bool matches(const ResourceAddress& address) const; +}; + +struct ResourceSnapshot { + ResourceAddress resource; + quint64 revision{0}; + QJsonObject value; + + [[nodiscard]] QJsonObject toJson() const; +}; + +// Main-thread authoritative cache for the bounded resources exposed by the +// AetherD control protocol. Revisions advance when the complete canonical value +// changes or a live identity is removed, so consumers never have to merge +// partial model state. +class ControlResourceStore final : public QObject { + Q_OBJECT + +public: + explicit ControlResourceStore(QObject* parent = nullptr); + + bool upsert(const ResourceAddress& address, const QJsonObject& value); + bool remove(const ResourceAddress& address); + + [[nodiscard]] std::optional get( + const ResourceAddress& address) const; + [[nodiscard]] QList snapshot( + const QList& selectors) const; + +signals: + void resourceChanged(const AetherSDR::control::ResourceSnapshot& snapshot); + void resourceRemoved(const AetherSDR::control::ResourceAddress& address, + quint64 revision); + +private: + QMap m_resources; + // Store-wide revisions are stronger than the protocol's per-identity + // monotonicity guarantee and avoid retaining one tombstone for every + // resource identity ever observed by a long-running daemon. + quint64 m_lastRevision{0}; +}; + +} // namespace AetherSDR::control diff --git a/src/core/control/ControlService.cpp b/src/core/control/ControlService.cpp index 203384138..9092a47cd 100644 --- a/src/core/control/ControlService.cpp +++ b/src/core/control/ControlService.cpp @@ -7,9 +7,102 @@ #include namespace AetherSDR::control { +namespace { + +constexpr qsizetype kMaxSelectorsPerSubscription = 64; + +std::optional onlyKeys( + const QJsonObject& object, const QSet& allowed) +{ + for (auto it = object.constBegin(); it != object.constEnd(); ++it) { + if (!allowed.contains(it.key())) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("unknown parameter"), + {{QStringLiteral("field"), it.key()}}, false}; + } + } + return std::nullopt; +} + +std::optional parseSelector( + const QJsonValue& value, bool wildcardAllowed, ResourceSelector* selector) +{ + if (!selector || !value.isObject()) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("resource selector must be an object"), {}, false}; + } + const QJsonObject object = value.toObject(); + if (const std::optional keyError = onlyKeys( + object, {QStringLiteral("type"), QStringLiteral("radioSession"), + QStringLiteral("id")})) { + return keyError; + } + + const QJsonValue typeValue = object.value(QStringLiteral("type")); + if (!typeValue.isString()) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("resource type must be a string"), {}, false}; + } + const QString type = typeValue.toString(); + const QSet supportedTypes{ + QStringLiteral("server"), QStringLiteral("radioSession"), + QStringLiteral("slice"), QStringLiteral("panadapter")}; + if (!supportedTypes.contains(type)) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("unsupported resource type"), + {{QStringLiteral("type"), type}}, false}; + } + + const QJsonValue radioSessionValue = object.value(QStringLiteral("radioSession")); + const QJsonValue idValue = object.value(QStringLiteral("id")); + const auto validOptionalId = [](const QJsonValue& field) { + return field.isUndefined() + || (field.isString() && !field.toString().isEmpty() + && field.toString().size() <= ProtocolLimits::kMaxRequestIdChars); + }; + if (!validOptionalId(radioSessionValue) || !validOptionalId(idValue)) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("resource identifiers must be bounded non-empty strings"), + {}, false}; + } + + const QString radioSession = radioSessionValue.toString(); + const QString id = idValue.toString(); + if (type == QStringLiteral("server")) { + if (!radioSessionValue.isUndefined() || !idValue.isUndefined()) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("server selector takes only type"), {}, false}; + } + } else if (type == QStringLiteral("radioSession")) { + if (!radioSessionValue.isUndefined() || (!wildcardAllowed && id.isEmpty())) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("radioSession selector requires id only"), {}, false}; + } + } else if (radioSession.isEmpty() || (!wildcardAllowed && id.isEmpty())) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("slice and panadapter selectors require radioSession and id"), + {}, false}; + } + + *selector = ResourceSelector{type, radioSession, id}; + return std::nullopt; +} + +ResourceAddress exactAddress(const ResourceSelector& selector) +{ + return {selector.type, selector.radioSession, selector.id}; +} + +} // namespace + +ControlService::ControlService(ControlResourceStore* resources) + : m_resources(resources) +{ + Q_ASSERT(m_resources); +} ServiceReply ControlService::handle( - const QByteArray& bytes, ControlSessionState* session) const + const QByteArray& bytes, ControlSession* session) const { const ParseResult parsed = ControlProtocolCodec::parseRequest(bytes); if (!parsed.ok()) { @@ -65,6 +158,77 @@ ServiceReply ControlService::handle( return {ControlProtocolCodec::successResponse( request.id, capabilities(*session)), false}; } + if (request.method == QStringLiteral("resource.get")) { + if (const std::optional keyError = onlyKeys( + request.params, {QStringLiteral("resource")})) { + return failure(request.id, *keyError); + } + ResourceSelector selector; + if (const std::optional selectorError = parseSelector( + request.params.value(QStringLiteral("resource")), false, &selector)) { + return failure(request.id, *selectorError); + } + const std::optional snapshot = + m_resources->get(exactAddress(selector)); + if (!snapshot) { + return failure(request.id, + {QStringLiteral("resource.not_found"), + QStringLiteral("resource does not exist"), {}, false}); + } + return {ControlProtocolCodec::successResponse(request.id, snapshot->toJson()), false}; + } + if (request.method == QStringLiteral("resource.subscribe")) { + if (const std::optional keyError = onlyKeys( + request.params, {QStringLiteral("resources")})) { + return failure(request.id, *keyError); + } + const QJsonValue resourcesValue = request.params.value(QStringLiteral("resources")); + if (!resourcesValue.isArray() || resourcesValue.toArray().isEmpty() + || resourcesValue.toArray().size() > kMaxSelectorsPerSubscription) { + return failure(request.id, + {QStringLiteral("request.invalid_params"), + QStringLiteral("resources must contain between 1 and 64 selectors"), + {}, false}); + } + QList selectors; + const QJsonArray resourceArray = resourcesValue.toArray(); + selectors.reserve(resourceArray.size()); + for (const QJsonValue& value : resourceArray) { + ResourceSelector selector; + if (const std::optional selectorError = + parseSelector(value, true, &selector)) { + return failure(request.id, *selectorError); + } + selectors.append(selector); + } + QJsonObject result; + if (const std::optional subscribeError = + session->subscribe(selectors, &result)) { + return failure(request.id, *subscribeError); + } + return {ControlProtocolCodec::successResponse(request.id, result), false}; + } + if (request.method == QStringLiteral("resource.unsubscribe")) { + if (const std::optional keyError = onlyKeys( + request.params, {QStringLiteral("subscription")})) { + return failure(request.id, *keyError); + } + const QJsonValue subscriptionValue = + request.params.value(QStringLiteral("subscription")); + if (!subscriptionValue.isString() || subscriptionValue.toString().isEmpty() + || subscriptionValue.toString().size() > ProtocolLimits::kMaxRequestIdChars) { + return failure(request.id, + {QStringLiteral("request.invalid_params"), + QStringLiteral("subscription must be a bounded non-empty string"), + {}, false}); + } + QJsonObject result; + if (const std::optional unsubscribeError = + session->unsubscribe(subscriptionValue.toString(), &result)) { + return failure(request.id, *unsubscribeError); + } + return {ControlProtocolCodec::successResponse(request.id, result), false}; + } return failure(request.id, {QStringLiteral("request.unknown_method"), @@ -72,7 +236,7 @@ ServiceReply ControlService::handle( {{QStringLiteral("method"), request.method}}, false}); } -QJsonObject ControlService::capabilities(const ControlSessionState& session) +QJsonObject ControlService::capabilities(const ControlSession& session) const { return { {QStringLiteral("sessionId"), session.sessionId}, @@ -81,9 +245,21 @@ QJsonObject ControlService::capabilities(const ControlSessionState& session) {QStringLiteral("name"), QStringLiteral("aetherd")}, {QStringLiteral("version"), QStringLiteral(AETHERSDR_VERSION)}}}, {QStringLiteral("grants"), QJsonArray{QStringLiteral("observe")}}, - {QStringLiteral("capabilities"), QJsonArray{QStringLiteral("server.read")}}, + {QStringLiteral("capabilities"), QJsonArray{ + QStringLiteral("server.read"), + QStringLiteral("radioSession.read"), + QStringLiteral("slice.read"), + QStringLiteral("panadapter.read"), + QStringLiteral("resource.get"), + QStringLiteral("resource.subscribe"), + QStringLiteral("resource.unsubscribe")}}, {QStringLiteral("limits"), QJsonObject{ - {QStringLiteral("maxMessageBytes"), ProtocolLimits::kMaxMessageBytes}}} + {QStringLiteral("maxMessageBytes"), ProtocolLimits::kMaxMessageBytes}, + {QStringLiteral("maxSubscriptions"), ControlSession::kMaxSubscriptions}, + {QStringLiteral("maxSelectorsPerSubscription"), + kMaxSelectorsPerSubscription}, + {QStringLiteral("maxQueuedOutputBytes"), + session.maxQueuedOutputBytes()}}} }; } diff --git a/src/core/control/ControlService.h b/src/core/control/ControlService.h index 4542d908d..b1d9a1c0f 100644 --- a/src/core/control/ControlService.h +++ b/src/core/control/ControlService.h @@ -1,31 +1,30 @@ #pragma once #include "ControlProtocolCodec.h" +#include "ControlResourceStore.h" +#include "ControlSession.h" #include #include namespace AetherSDR::control { -struct ControlSessionState { - QString sessionId; - bool negotiated{false}; -}; - struct ServiceReply { QJsonObject message; bool closeAfterWrite{false}; }; -// Transport-neutral Stage-3 service kernel. The first landed surface is -// intentionally observe-only: negotiation and capability discovery. Model -// resources and non-TX control methods attach here in subsequent slices. +// Transport-neutral Stage-3 service kernel. The current surface is strictly +// observe-only: negotiation, capability discovery, typed resource reads, and +// subscriptions. Non-TX control methods attach in a subsequent slice. class ControlService final { public: + explicit ControlService(ControlResourceStore* resources); + [[nodiscard]] ServiceReply handle( - const QByteArray& bytes, ControlSessionState* session) const; + const QByteArray& bytes, ControlSession* session) const; - [[nodiscard]] static QJsonObject capabilities(const ControlSessionState& session); + [[nodiscard]] QJsonObject capabilities(const ControlSession& session) const; private: [[nodiscard]] static ServiceReply failure( @@ -33,6 +32,8 @@ class ControlService final { [[nodiscard]] static std::optional validateHelloParams( const QJsonObject& params); [[nodiscard]] static bool acceptsVersionOne(const QJsonObject& params); + + ControlResourceStore* m_resources{nullptr}; }; } // namespace AetherSDR::control diff --git a/src/core/control/ControlSession.cpp b/src/core/control/ControlSession.cpp new file mode 100644 index 000000000..db64b5994 --- /dev/null +++ b/src/core/control/ControlSession.cpp @@ -0,0 +1,219 @@ +#include "ControlSession.h" + +#include +#include + +#include + +namespace AetherSDR::control { + +ControlSession::ControlSession(ControlResourceStore* resources, + qint64 maxQueuedOutputBytes, + QObject* parent) + : QObject(parent), + m_resources(resources), + m_maxQueuedOutputBytes(maxQueuedOutputBytes) +{ + Q_ASSERT(m_resources); + connect(m_resources, &ControlResourceStore::resourceChanged, + this, &ControlSession::onResourceChanged); + connect(m_resources, &ControlResourceStore::resourceRemoved, + this, &ControlSession::onResourceRemoved); +} + +std::optional ControlSession::subscribe( + const QList& selectors, QJsonObject* result) +{ + if (!result || selectors.isEmpty()) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("resources must be a non-empty array"), {}, false}; + } + if (m_subscriptions.size() >= kMaxSubscriptions) { + return ProtocolError{QStringLiteral("transport.limit_exceeded"), + QStringLiteral("maximum subscriptions reached"), {}, false}; + } + + if (m_resyncRequired) { + // A fresh atomic baseline supersedes an undrained resync notice. Do not + // deliver that older invalidation after the successful subscribe reply. + m_pending.clear(); + m_pendingBytes = 0; + } + const QString subscriptionId = QStringLiteral("sub-%1").arg(m_nextSubscription++); + m_subscriptions.insert(subscriptionId, selectors); + rebuildSelectorIndex(); + m_resyncRequired = false; + + QJsonArray resources; + const QList snapshots = m_resources->snapshot(selectors); + for (const ResourceSnapshot& snapshot : snapshots) { + resources.append(snapshot.toJson()); + } + *result = {{QStringLiteral("subscription"), subscriptionId}, + {QStringLiteral("sequence"), static_cast(m_drainedSequence)}, + {QStringLiteral("resources"), resources}}; + return std::nullopt; +} + +std::optional ControlSession::unsubscribe( + const QString& subscriptionId, QJsonObject* result) +{ + if (!result || subscriptionId.isEmpty()) { + return ProtocolError{QStringLiteral("request.invalid_params"), + QStringLiteral("subscription must be a non-empty string"), + {}, false}; + } + if (m_subscriptions.remove(subscriptionId) == 0) { + return ProtocolError{QStringLiteral("resource.not_found"), + QStringLiteral("subscription does not exist"), {}, false}; + } + rebuildSelectorIndex(); + for (qsizetype index = m_pending.size(); index > 0; --index) { + const PendingMessage& pending = m_pending.at(index - 1); + if (pending.resource && !observes(*pending.resource)) { + m_pendingBytes -= pending.frame.size(); + m_pending.removeAt(index - 1); + } + } + *result = {{QStringLiteral("subscription"), subscriptionId}, + {QStringLiteral("removed"), true}}; + return std::nullopt; +} + +QList ControlSession::takePendingFrames() +{ + QList frames; + frames.reserve(m_pending.size()); + for (const PendingMessage& pending : std::as_const(m_pending)) { + frames.append(pending.frame); + if (pending.sequence > m_drainedSequence) { + m_drainedSequence = pending.sequence; + } + } + m_pending.clear(); + m_pendingBytes = 0; + return frames; +} + +void ControlSession::rebuildSelectorIndex() +{ + m_selectorsByType.clear(); + for (auto it = m_subscriptions.constBegin(); it != m_subscriptions.constEnd(); ++it) { + for (const ResourceSelector& selector : it.value()) { + m_selectorsByType[selector.type].append(selector); + } + } +} + +bool ControlSession::observes(const ResourceAddress& address) const +{ + const auto bucket = m_selectorsByType.constFind(address.type); + if (bucket == m_selectorsByType.constEnd()) { + return false; + } + for (const ResourceSelector& selector : *bucket) { + if (selector.matches(address)) { + return true; + } + } + return false; +} + +void ControlSession::onResourceChanged(const ResourceSnapshot& snapshot) +{ + if (!m_resyncRequired && observes(snapshot.resource)) { + enqueueResourceEvent(QStringLiteral("resource.changed"), + snapshot.resource, snapshot.revision, snapshot.value); + } +} + +void ControlSession::onResourceRemoved( + const ResourceAddress& address, quint64 revision) +{ + if (!m_resyncRequired && observes(address)) { + enqueueResourceEvent(QStringLiteral("resource.removed"), address, revision); + } +} + +void ControlSession::enqueueResourceEvent( + const QString& event, const ResourceAddress& address, + quint64 revision, const QJsonObject& value) +{ + QJsonObject message{{QStringLiteral("v"), 1}, + {QStringLiteral("sessionId"), sessionId}, + {QStringLiteral("event"), event}, + {QStringLiteral("sequence"), static_cast(++m_sequence)}, + {QStringLiteral("resource"), address.toJson()}, + {QStringLiteral("revision"), static_cast(revision)}}; + if (event == QStringLiteral("resource.changed")) { + message.insert(QStringLiteral("value"), value); + } + enqueueCoalesced(address, message); +} + +void ControlSession::enqueueCoalesced( + const ResourceAddress& address, const QJsonObject& message) +{ + const QString key = address.key(); + const QByteArray frame = encodeFrame(message); + const qint64 bytes = frame.size(); + const quint64 sequence = m_sequence; + for (qsizetype index = 0; index < m_pending.size(); ++index) { + const PendingMessage& pending = m_pending.at(index); + if (pending.coalesceKey == key) { + const qint64 nextBytes = m_pendingBytes - pending.frame.size() + bytes; + if (nextBytes > m_maxQueuedOutputBytes) { + requireResync(); + return; + } + m_pendingBytes = nextBytes; + m_pending.removeAt(index); + m_pending.append(PendingMessage{key, address, frame, sequence}); + emit outputReady(); + return; + } + } + + if (m_pendingBytes + bytes > m_maxQueuedOutputBytes) { + requireResync(); + return; + } + m_pending.append(PendingMessage{key, address, frame, sequence}); + m_pendingBytes += bytes; + emit outputReady(); +} + +void ControlSession::requireResync() +{ + if (m_resyncRequired) { + return; + } + m_resyncRequired = true; + m_subscriptions.clear(); + rebuildSelectorIndex(); + m_pending.clear(); + m_pendingBytes = 0; + + const QJsonObject message{{QStringLiteral("v"), 1}, + {QStringLiteral("sessionId"), sessionId}, + {QStringLiteral("event"), QStringLiteral("resource.resyncRequired")}, + {QStringLiteral("sequence"), static_cast(++m_sequence)}, + {QStringLiteral("subscriptionsInvalidated"), true}}; + const QByteArray frame = encodeFrame(message); + if (frame.size() > m_maxQueuedOutputBytes) { + emit outputOverflow(); + return; + } + m_pending.append(PendingMessage{QString(), std::nullopt, frame, m_sequence}); + m_pendingBytes = frame.size(); + emit outputReady(); +} + +QByteArray ControlSession::encodeFrame(const QJsonObject& message) +{ + QByteArray frame = QJsonDocument(message).toJson(QJsonDocument::Compact); + frame.append('\n'); + return frame; +} + +} // namespace AetherSDR::control diff --git a/src/core/control/ControlSession.h b/src/core/control/ControlSession.h new file mode 100644 index 000000000..4b03d27d5 --- /dev/null +++ b/src/core/control/ControlSession.h @@ -0,0 +1,85 @@ +#pragma once + +#include "ControlProtocolCodec.h" +#include "ControlResourceStore.h" + +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace AetherSDR::control { + +// Per-client protocol state. Resource events are session-sequenced and held in +// a bounded, coalescing queue until the transport drains them. +class ControlSession final : public QObject { + Q_OBJECT + +public: + static constexpr int kMaxSubscriptions = 64; + + explicit ControlSession(ControlResourceStore* resources, + qint64 maxQueuedOutputBytes, + QObject* parent = nullptr); + + QString sessionId; + bool negotiated{false}; + + [[nodiscard]] std::optional subscribe( + const QList& selectors, QJsonObject* result); + [[nodiscard]] std::optional unsubscribe( + const QString& subscriptionId, QJsonObject* result); + // Complete wire frames (compact JSON plus the framing newline), encoded + // once at enqueue time so neither the queue bound nor the transport has to + // serialize the message again. + [[nodiscard]] QList takePendingFrames(); + [[nodiscard]] quint64 sequence() const { return m_sequence; } + [[nodiscard]] qint64 maxQueuedOutputBytes() const { return m_maxQueuedOutputBytes; } + +signals: + void outputReady(); + void outputOverflow(); + +private: + struct PendingMessage { + QString coalesceKey; + std::optional resource; + QByteArray frame; + quint64 sequence{0}; + }; + + [[nodiscard]] bool observes(const ResourceAddress& address) const; + void onResourceChanged(const ResourceSnapshot& snapshot); + void onResourceRemoved(const ResourceAddress& address, quint64 revision); + void enqueueResourceEvent(const QString& event, + const ResourceAddress& address, + quint64 revision, + const QJsonObject& value = {}); + void enqueueCoalesced(const ResourceAddress& address, + const QJsonObject& message); + void requireResync(); + void rebuildSelectorIndex(); + [[nodiscard]] static QByteArray encodeFrame(const QJsonObject& message); + + ControlResourceStore* m_resources{nullptr}; + qint64 m_maxQueuedOutputBytes{0}; + quint64 m_sequence{0}; + quint64 m_drainedSequence{0}; + quint64 m_nextSubscription{1}; + QMap> m_subscriptions; + // Derived from m_subscriptions on every subscription change. observes() runs + // once per session per store change, so scanning all 64x64 advertised + // selectors there would put that sweep on the model's publish path; bucketed + // by resource type it only ever visits selectors that could match. + QHash> m_selectorsByType; + QList m_pending; + qint64 m_pendingBytes{0}; + bool m_resyncRequired{false}; +}; + +} // namespace AetherSDR::control diff --git a/src/core/control/LocalControlServer.cpp b/src/core/control/LocalControlServer.cpp index d61a1c7ba..683a419b5 100644 --- a/src/core/control/LocalControlServer.cpp +++ b/src/core/control/LocalControlServer.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -16,10 +17,27 @@ #endif namespace AetherSDR::control { +namespace { + +QJsonObject serverValue(const QString& localTransport) +{ + return {{QStringLiteral("name"), QStringLiteral("aetherd")}, + {QStringLiteral("buildVersion"), QStringLiteral(AETHERSDR_VERSION)}, + {QStringLiteral("protocolVersions"), QJsonArray{1}}, + {QStringLiteral("health"), QStringLiteral("ok")}, + {QStringLiteral("localTransport"), localTransport}}; +} + +} // namespace struct LocalControlServer::Client { + Client(ControlResourceStore* resources, qint64 maxQueuedOutputBytes) + : session(std::make_unique(resources, maxQueuedOutputBytes)) + { + } + QByteArray input; - ControlSessionState session; + std::unique_ptr session; QTimer handshakeTimer; }; @@ -29,8 +47,11 @@ LocalControlServer::LocalControlServer(QObject* parent) } LocalControlServer::LocalControlServer(QObject* parent, Limits limits) - : QObject(parent), m_limits(limits) + : QObject(parent), m_resources(), m_service(&m_resources), m_limits(limits) { + m_resources.upsert( + {QStringLiteral("server"), {}, {}}, + serverValue(QStringLiteral("idle"))); m_server.setSocketOptions(QLocalServer::UserAccessOption); connect(&m_server, &QLocalServer::newConnection, this, [this] { acceptConnections(); }); @@ -78,11 +99,15 @@ bool LocalControlServer::listen(const QString& name) return false; } m_lock = std::move(lock); + m_resources.upsert( + {QStringLiteral("server"), {}, {}}, + serverValue(QStringLiteral("listening"))); return true; } void LocalControlServer::close() { + const bool wasListening = m_server.isListening(); m_server.close(); QList sockets; sockets.reserve(static_cast(m_clients.size())); @@ -95,6 +120,11 @@ void LocalControlServer::close() dropClient(socket); } m_lock.reset(); + if (wasListening) { + m_resources.upsert( + {QStringLiteral("server"), {}, {}}, + serverValue(QStringLiteral("stopped"))); + } } void LocalControlServer::acceptConnections() @@ -115,7 +145,8 @@ void LocalControlServer::acceptConnections() continue; } - std::unique_ptr ownedClient = std::make_unique(); + std::unique_ptr ownedClient = + std::make_unique(&m_resources, m_limits.maxQueuedOutputBytes); Client* client = ownedClient.get(); client->handshakeTimer.setSingleShot(true); client->handshakeTimer.setInterval(m_limits.handshakeTimeoutMs); @@ -132,6 +163,11 @@ void LocalControlServer::acceptConnections() }); connect(socket, &QLocalSocket::readyRead, this, [this, socket] { readClient(socket); }); + connect(client->session.get(), &ControlSession::outputReady, + this, [this, socket] { drainSessionOutput(socket); }, + Qt::QueuedConnection); + connect(client->session.get(), &ControlSession::outputOverflow, + socket, &QLocalSocket::abort, Qt::QueuedConnection); connect(socket, &QLocalSocket::disconnected, this, [this, socket] { // QLocalSocket::abort() may emit disconnected synchronously @@ -176,8 +212,8 @@ void LocalControlServer::readClient(QLocalSocket* socket) frame.chop(1); } - const ServiceReply reply = m_service.handle(frame, &client->session); - if (client->session.negotiated) { + const ServiceReply reply = m_service.handle(frame, client->session.get()); + if (client->session->negotiated) { client->handshakeTimer.stop(); } if (!send(socket, reply.message)) { @@ -190,23 +226,42 @@ void LocalControlServer::readClient(QLocalSocket* socket) } } +void LocalControlServer::drainSessionOutput(QLocalSocket* socket) +{ + const auto clientIt = m_clients.find(socket); + if (clientIt == m_clients.end()) { + return; + } + const QList frames = clientIt->second->session->takePendingFrames(); + for (const QByteArray& frame : frames) { + if (!sendFrame(socket, frame)) { + return; + } + } +} + void LocalControlServer::dropClient(QLocalSocket* socket) { m_clients.erase(socket); } bool LocalControlServer::send(QLocalSocket* socket, const QJsonObject& message) +{ + QByteArray bytes = QJsonDocument(message).toJson(QJsonDocument::Compact); + bytes.append('\n'); + return sendFrame(socket, bytes); +} + +bool LocalControlServer::sendFrame(QLocalSocket* socket, const QByteArray& frame) { if (!socket || socket->state() == QLocalSocket::UnconnectedState) { return false; } - QByteArray bytes = QJsonDocument(message).toJson(QJsonDocument::Compact); - bytes.append('\n'); - if (socket->bytesToWrite() + bytes.size() > m_limits.maxQueuedOutputBytes) { + if (socket->bytesToWrite() + frame.size() > m_limits.maxQueuedOutputBytes) { socket->abort(); return false; } - if (socket->write(bytes) != bytes.size()) { + if (socket->write(frame) != frame.size()) { socket->abort(); return false; } diff --git a/src/core/control/LocalControlServer.h b/src/core/control/LocalControlServer.h index 6dcb21c07..2adb0da46 100644 --- a/src/core/control/LocalControlServer.h +++ b/src/core/control/LocalControlServer.h @@ -37,18 +37,24 @@ class LocalControlServer final : public QObject { void close(); [[nodiscard]] bool isListening() const { return m_server.isListening(); } [[nodiscard]] QString fullServerName() const { return m_server.fullServerName(); } + [[nodiscard]] ControlResourceStore& resourceStore() { return m_resources; } private: struct Client; void acceptConnections(); void readClient(QLocalSocket* socket); + void drainSessionOutput(QLocalSocket* socket); void dropClient(QLocalSocket* socket); [[nodiscard]] bool send(QLocalSocket* socket, const QJsonObject& message); + // Writes a frame the session already encoded, so an event is serialized + // exactly once between ControlSession and the socket. + [[nodiscard]] bool sendFrame(QLocalSocket* socket, const QByteArray& frame); [[nodiscard]] static bool resolveEndpoint( const QString& logicalName, QString* endpointName, QString* lockPath); QLocalServer m_server; + ControlResourceStore m_resources; ControlService m_service; Limits m_limits; std::unordered_map> m_clients; diff --git a/src/core/control/RadioResourceAdapter.cpp b/src/core/control/RadioResourceAdapter.cpp new file mode 100644 index 000000000..f56d487d0 --- /dev/null +++ b/src/core/control/RadioResourceAdapter.cpp @@ -0,0 +1,348 @@ +#include "RadioResourceAdapter.h" + +#include "core/backends/RadioCapabilities.h" +#include "models/PanadapterModel.h" +#include "models/RadioModel.h" +#include "models/SliceModel.h" + +#include +#include + +#include + +namespace AetherSDR::control { +namespace { + +QJsonArray strings(const auto& values) +{ + QJsonArray result; + for (const auto& value : values) { + result.append(value); + } + return result; +} + +QJsonObject capabilityValue(const RadioCapabilities& capabilities) +{ + QJsonArray sampleRates; + for (int sampleRate : capabilities.sampleRatesHz) { + sampleRates.append(sampleRate); + } + QJsonArray bands; + for (const DeclaredBandRange& band : capabilities.declaredBandRanges) { + bands.append(QJsonObject{{QStringLiteral("name"), band.name}, + {QStringLiteral("lowHz"), band.lowHz}, + {QStringLiteral("highHz"), band.highHz}}); + } + return { + {QStringLiteral("maxSlices"), capabilities.maxSlices}, + {QStringLiteral("maxPanadapters"), capabilities.maxPanadapters}, + {QStringLiteral("sampleRatesHz"), sampleRates}, + {QStringLiteral("tuningRangeHz"), QJsonObject{ + {QStringLiteral("minimum"), capabilities.tuningMinHz}, + {QStringLiteral("maximum"), capabilities.tuningMaxHz}}}, + {QStringLiteral("declaredBands"), bands}, + {QStringLiteral("canTransmit"), capabilities.canTransmit}, + {QStringLiteral("maximumTransmitWatts"), capabilities.txPowerMaxWatts}, + {QStringLiteral("hasTuner"), capabilities.hasTuner}, + {QStringLiteral("hasAmplifier"), capabilities.hasAmplifier}, + {QStringLiteral("extensions"), strings(capabilities.extensionNamespaces)} + }; +} + +} // namespace + +RadioResourceAdapter::RadioResourceAdapter( + RadioModel* radio, ControlResourceStore* resources, + QString radioSessionId, QObject* parent) + : QObject(parent), + m_radio(radio), + m_resources(resources), + m_radioSessionId(std::move(radioSessionId)) +{ + Q_ASSERT(m_radio); + Q_ASSERT(m_resources); + Q_ASSERT(!m_radioSessionId.isEmpty()); + + connect(m_radio, &RadioModel::infoChanged, + this, &RadioResourceAdapter::publishRadioSession); + connect(m_radio, &RadioModel::connectionStateChanged, + this, [this](bool connected) { + publishRadioSession(); + if (connected) { + publishAll(); + } else { + clearDynamicResources(); + } + }); + connect(m_radio, &RadioModel::capabilitiesChanged, + this, [this] { publishRadioSession(); }); + connect(m_radio, &RadioModel::backendRebuilt, + this, [this] { publishRadioSession(); }); + + connect(m_radio, &RadioModel::sliceAdded, + this, &RadioResourceAdapter::attachSlice); + connect(m_radio, &RadioModel::sliceRemoved, this, [this](int sliceId) { + // Match the object, not just the id. pruneStaleSessionModels() removes a + // *stale* SliceModel whose id a live one may already have reclaimed; an + // id-only match would unbind the live slice and publish a spurious + // resource.removed for a resource that still exists. + SliceModel* live = m_radio->slice(sliceId); + SliceModel* removed = nullptr; + for (SliceModel* slice : std::as_const(m_slices)) { + if (slice && slice != live && slice->sliceId() == sliceId) { + removed = slice; + break; + } + } + if (!removed) { + return; + } + QObject::disconnect(removed, nullptr, this, nullptr); + m_slices.remove(removed); + m_resources->remove({QStringLiteral("slice"), m_radioSessionId, + QString::number(sliceId)}); + }); + connect(m_radio, &RadioModel::slotOccupancyChanged, this, [this](int sliceId) { + refreshSlice(m_radio->slice(sliceId)); + }); + + connect(m_radio, &RadioModel::panadapterAdded, + this, &RadioResourceAdapter::attachPanadapter); + connect(m_radio, &RadioModel::panadapterReclaimed, + this, &RadioResourceAdapter::refreshPanadapter); + connect(m_radio, &RadioModel::panadapterRemoved, + this, [this](const QString& panId) { + PanadapterModel* removed = nullptr; + for (PanadapterModel* panadapter : std::as_const(m_panadapters)) { + if (panadapter && panadapter->panId() == panId) { + removed = panadapter; + break; + } + } + if (removed) { + QObject::disconnect(removed, nullptr, this, nullptr); + } + m_panadapters.remove(removed); + m_resources->remove({QStringLiteral("panadapter"), m_radioSessionId, + panId}); + }); + + publishAll(); +} + +void RadioResourceAdapter::publishAll() +{ + publishRadioSession(); + for (SliceModel* slice : m_radio->slices()) { + attachSlice(slice); + } + for (PanadapterModel* panadapter : m_radio->panadapters()) { + attachPanadapter(panadapter); + } +} + +void RadioResourceAdapter::attachSlice(SliceModel* slice) +{ + if (!slice || m_slices.contains(slice)) { + return; + } + m_slices.insert(slice); + const auto refresh = [this, slice] { publishSlice(slice); }; + connect(slice, &SliceModel::letterChanged, this, refresh); + connect(slice, &SliceModel::frequencyChanged, this, refresh); + connect(slice, &SliceModel::panIdChanged, this, refresh); + connect(slice, &SliceModel::modeChanged, this, refresh); + connect(slice, &SliceModel::filterChanged, this, refresh); + connect(slice, &SliceModel::activeChanged, this, refresh); + connect(slice, &SliceModel::txSliceChanged, this, refresh); + connect(slice, &SliceModel::audioGainChanged, this, refresh); + connect(slice, &SliceModel::audioPanChanged, this, refresh); + connect(slice, &SliceModel::audioMuteChanged, this, refresh); + connect(slice, &SliceModel::rxAntennaChanged, this, refresh); + connect(slice, &SliceModel::lockedChanged, this, refresh); + connect(slice, &SliceModel::rfGainChanged, this, refresh); + connect(slice, &SliceModel::agcModeChanged, this, refresh); + connect(slice, &SliceModel::agcThresholdChanged, this, refresh); + connect(slice, &SliceModel::agcOffLevelChanged, this, refresh); + connect(slice, &SliceModel::squelchChanged, this, refresh); + // While external receive-audio replacement is active (a Kiwi RX source + // replacing the radio's audio) SliceModel's AGC and squelch setters take an + // early-return branch that emits only these signals, but receiveAgcMode(), + // receiveAgcThreshold(), receiveAgcOffLevel(), receiveSquelchOn() and + // receiveSquelchLevel() switch to the external values. Without them + // receive.agc.* and receive.squelch.* would publish the pre-replacement + // state forever. (audio.* needs no equivalent: setAudioGain/Pan/Mute reuse + // the same signal in both branches.) + connect(slice, &SliceModel::externalReceiveAgcModeChanged, this, refresh); + connect(slice, &SliceModel::externalReceiveAgcThresholdChanged, this, refresh); + connect(slice, &SliceModel::externalReceiveAgcOffLevelChanged, this, refresh); + connect(slice, &SliceModel::externalReceiveSquelchChanged, this, refresh); + connect(slice, &QObject::destroyed, this, [this, slice] { + m_slices.remove(slice); + }); + publishSlice(slice); +} + +void RadioResourceAdapter::refreshSlice(SliceModel* slice) +{ + if (!slice) { + return; + } + if (!m_slices.contains(slice)) { + attachSlice(slice); + return; + } + publishSlice(slice); +} + +void RadioResourceAdapter::attachPanadapter(PanadapterModel* panadapter) +{ + if (!panadapter || m_panadapters.contains(panadapter)) { + return; + } + m_panadapters.insert(panadapter); + const auto refresh = [this, panadapter] { publishPanadapter(panadapter); }; + connect(panadapter, &PanadapterModel::infoChanged, this, refresh); + connect(panadapter, &PanadapterModel::levelChanged, this, refresh); + connect(panadapter, &PanadapterModel::bandwidthLimitsChanged, this, refresh); + connect(panadapter, &PanadapterModel::rxAntennaChanged, this, refresh); + connect(panadapter, &PanadapterModel::rfGainChanged, this, refresh); + connect(panadapter, &PanadapterModel::fpsChanged, this, refresh); + connect(panadapter, &PanadapterModel::averageChanged, this, refresh); + // The first valid report can be false, matching the model's default value. + // Listen to Reported so weightedAverageKnown is published on that edge; + // ControlResourceStore deduplicates later identical snapshots. + connect(panadapter, &PanadapterModel::weightedAverageReported, this, refresh); + connect(panadapter, &PanadapterModel::waterfallLineDurationChanged, this, refresh); + connect(panadapter, &QObject::destroyed, this, [this, panadapter] { + m_panadapters.remove(panadapter); + }); + publishPanadapter(panadapter); +} + +void RadioResourceAdapter::refreshPanadapter(PanadapterModel* panadapter) +{ + if (!panadapter) { + return; + } + if (!m_panadapters.contains(panadapter)) { + attachPanadapter(panadapter); + return; + } + publishPanadapter(panadapter); +} + +void RadioResourceAdapter::clearDynamicResources() +{ + const QSet slices = m_slices; + m_slices.clear(); + for (SliceModel* slice : slices) { + if (!slice) { + continue; + } + QObject::disconnect(slice, nullptr, this, nullptr); + m_resources->remove({QStringLiteral("slice"), m_radioSessionId, + QString::number(slice->sliceId())}); + } + + const QSet panadapters = m_panadapters; + m_panadapters.clear(); + for (PanadapterModel* panadapter : panadapters) { + if (!panadapter) { + continue; + } + QObject::disconnect(panadapter, nullptr, this, nullptr); + m_resources->remove({QStringLiteral("panadapter"), m_radioSessionId, + panadapter->panId()}); + } +} + +void RadioResourceAdapter::publishRadioSession() +{ + const RadioCapabilities capabilities = m_radio->backendCapabilities(); + const QJsonObject value{ + {QStringLiteral("id"), m_radioSessionId}, + {QStringLiteral("connected"), m_radio->isConnected()}, + {QStringLiteral("family"), m_radio->family()}, + {QStringLiteral("identity"), QJsonObject{ + {QStringLiteral("name"), m_radio->name()}, + {QStringLiteral("model"), m_radio->model()}, + {QStringLiteral("serial"), m_radio->serial()}, + {QStringLiteral("version"), m_radio->version()}, + {QStringLiteral("manufacturer"), capabilities.manufacturer}}}, + {QStringLiteral("capabilities"), capabilityValue(capabilities)}}; + m_resources->upsert({QStringLiteral("radioSession"), {}, m_radioSessionId}, value); +} + +void RadioResourceAdapter::publishSlice(SliceModel* slice) +{ + if (!slice || !m_slices.contains(slice)) { + return; + } + const QJsonObject value{ + {QStringLiteral("id"), QString::number(slice->sliceId())}, + {QStringLiteral("letter"), slice->letter()}, + {QStringLiteral("panadapterId"), slice->panId()}, + {QStringLiteral("owned"), m_radio->isSlotOurs(slice->sliceId())}, + {QStringLiteral("frequencyHz"), qRound64(slice->frequency() * 1'000'000.0)}, + {QStringLiteral("mode"), slice->mode()}, + {QStringLiteral("filter"), QJsonObject{ + {QStringLiteral("lowHz"), slice->filterLow()}, + {QStringLiteral("highHz"), slice->filterHigh()}}}, + {QStringLiteral("active"), slice->isActive()}, + {QStringLiteral("txSlice"), slice->isTxSlice()}, + {QStringLiteral("locked"), slice->isLocked()}, + {QStringLiteral("audio"), QJsonObject{ + {QStringLiteral("gain"), slice->audioGain()}, + {QStringLiteral("pan"), slice->audioPan()}, + {QStringLiteral("muted"), slice->audioMute()}}}, + {QStringLiteral("receive"), QJsonObject{ + {QStringLiteral("antenna"), slice->rxAntenna()}, + {QStringLiteral("rfGain"), slice->rfGain()}, + {QStringLiteral("agc"), QJsonObject{ + {QStringLiteral("mode"), slice->receiveAgcMode()}, + {QStringLiteral("threshold"), slice->receiveAgcThreshold()}, + {QStringLiteral("offLevel"), slice->receiveAgcOffLevel()}}}, + {QStringLiteral("squelch"), QJsonObject{ + {QStringLiteral("enabled"), slice->receiveSquelchOn()}, + {QStringLiteral("level"), slice->receiveSquelchLevel()}}}}}}; + m_resources->upsert({QStringLiteral("slice"), m_radioSessionId, + QString::number(slice->sliceId())}, value); +} + +void RadioResourceAdapter::publishPanadapter(PanadapterModel* panadapter) +{ + if (!panadapter || !m_panadapters.contains(panadapter)) { + return; + } + const QJsonObject value{ + {QStringLiteral("id"), panadapter->panId()}, + {QStringLiteral("centerHz"), qRound64(panadapter->centerMhz() * 1'000'000.0)}, + {QStringLiteral("centerKnown"), panadapter->centerKnown()}, + {QStringLiteral("bandwidthHz"), + qRound64(panadapter->bandwidthMhz() * 1'000'000.0)}, + {QStringLiteral("dbmRange"), QJsonObject{ + {QStringLiteral("minimum"), panadapter->minDbm()}, + {QStringLiteral("maximum"), panadapter->maxDbm()}}}, + {QStringLiteral("bandwidthLimitsHz"), QJsonObject{ + {QStringLiteral("minimum"), + qRound64(panadapter->minBandwidthMhz() * 1'000'000.0)}, + {QStringLiteral("maximum"), + qRound64(panadapter->maxBandwidthMhz() * 1'000'000.0)}}}, + {QStringLiteral("receive"), QJsonObject{ + {QStringLiteral("antenna"), panadapter->rxAntenna()}, + {QStringLiteral("rfGain"), panadapter->rfGain()}}}, + {QStringLiteral("displayCadence"), QJsonObject{ + {QStringLiteral("fps"), panadapter->fps()}, + {QStringLiteral("averageFrames"), panadapter->average()}, + {QStringLiteral("weightedAverage"), panadapter->weightedAverage()}, + {QStringLiteral("weightedAverageKnown"), + panadapter->weightedAverageKnown()}, + {QStringLiteral("waterfallRate"), + panadapter->waterfallLineDuration()}}}}; + m_resources->upsert({QStringLiteral("panadapter"), m_radioSessionId, + panadapter->panId()}, value); +} + +} // namespace AetherSDR::control diff --git a/src/core/control/RadioResourceAdapter.h b/src/core/control/RadioResourceAdapter.h new file mode 100644 index 000000000..82a823d9e --- /dev/null +++ b/src/core/control/RadioResourceAdapter.h @@ -0,0 +1,50 @@ +#pragma once + +#include "ControlResourceStore.h" + +#include +#include +#include + +namespace AetherSDR { + +class PanadapterModel; +class RadioModel; +class SliceModel; + +namespace control { + +// Converts the existing normalized model graph into complete protocol +// resources. It is strictly observational: no model setter or backend intent +// is reachable through this adapter. +class RadioResourceAdapter final : public QObject { + Q_OBJECT + +public: + RadioResourceAdapter(RadioModel* radio, + ControlResourceStore* resources, + QString radioSessionId, + QObject* parent = nullptr); + + [[nodiscard]] QString radioSessionId() const { return m_radioSessionId; } + void publishAll(); + +private: + void attachSlice(SliceModel* slice); + void refreshSlice(SliceModel* slice); + void attachPanadapter(PanadapterModel* panadapter); + void refreshPanadapter(PanadapterModel* panadapter); + void clearDynamicResources(); + void publishRadioSession(); + void publishSlice(SliceModel* slice); + void publishPanadapter(PanadapterModel* panadapter); + + RadioModel* m_radio{nullptr}; + ControlResourceStore* m_resources{nullptr}; + QString m_radioSessionId; + QSet m_slices; + QSet m_panadapters; +}; + +} // namespace control +} // namespace AetherSDR diff --git a/src/models/RadioModel.cpp b/src/models/RadioModel.cpp index c5c16dcce..1605a3f66 100644 --- a/src/models/RadioModel.cpp +++ b/src/models/RadioModel.cpp @@ -1289,6 +1289,11 @@ void RadioModel::setupBackend(const QString& family) s->applyChanges(mapped); m_meterModel.setActiveTxSlice(activeTxSliceNum()); refreshTxPowerLimit(); + // Reclaim deliberately does not emit sliceAdded: the UI already + // owns this object. Notify non-UI observers through the existing + // occupancy edge so adapters that detached on disconnect can + // reattach and republish it. + emit slotOccupancyChanged(sliceId); // Reuse the same SliceModel so every UI subscriber — including // RX Controls — stays attached. A sliceAdded here would build a // duplicate VFO for an object the UI already owns. diff --git a/src/models/RadioModel.h b/src/models/RadioModel.h index 782900743..4013a00b5 100644 --- a/src/models/RadioModel.h +++ b/src/models/RadioModel.h @@ -1615,6 +1615,19 @@ private slots: { onStatusReceived(object, kvs); } + // Drive the reconnect/reclaim portion of the normalized backend seam + // without a synthetic radio peer. The socket-free resource test uses these + // to prove that a reclaimed non-Flex slice is republished. + void stageSessionModelsForReconnectForTest() + { + stageSessionModelsForReconnect(); + } + void emitBackendSliceChangedForTest(int sliceId, const SliceDelta& delta) + { + if (m_backend) { + emit m_backend->sliceChanged(sliceId, delta); + } + } private: PanadapterModel* resolveBackendPan(const QString& backendPanId); diff --git a/tests/control_resource_service_test.cpp b/tests/control_resource_service_test.cpp new file mode 100644 index 000000000..5999286e6 --- /dev/null +++ b/tests/control_resource_service_test.cpp @@ -0,0 +1,868 @@ +#include "TestSettingsProfile.h" +#include "core/RadioDiscovery.h" +#include "core/backends/sim/SimBackend.h" +#include "core/control/ControlResourceStore.h" +#include "core/control/ControlService.h" +#include "core/control/ControlSession.h" +#include "core/control/RadioResourceAdapter.h" +#include "models/RadioModel.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +using namespace AetherSDR; +using namespace AetherSDR::control; + +namespace { + +bool check(bool condition, const char* message) +{ + if (condition) { + return true; + } + std::fprintf(stderr, "%s\n", message); + return false; +} + +bool hasExactlyKeys(const QJsonObject& object, + std::initializer_list expected) +{ + QSet expectedKeys; + for (const char* key : expected) { + expectedKeys.insert(QString::fromLatin1(key)); + } + QSet actualKeys; + for (auto it = object.constBegin(); it != object.constEnd(); ++it) { + actualKeys.insert(it.key()); + } + return actualKeys == expectedKeys; +} + +bool waitUntil(const std::function& predicate, int timeoutMs = 3000) +{ + QElapsedTimer timer; + timer.start(); + while (!predicate() && timer.elapsed() < timeoutMs) { + QCoreApplication::processEvents(QEventLoop::AllEvents, 10); + QThread::msleep(1); + } + return predicate(); +} + +QJsonObject invoke(ControlService* service, ControlSession* session, + const QString& id, const QString& method, + const QJsonObject& params) +{ + QJsonObject request{{QStringLiteral("v"), 1}, + {QStringLiteral("id"), id}, + {QStringLiteral("method"), method}, + {QStringLiteral("params"), params}}; + if (method != QStringLiteral("hello")) { + request.insert(QStringLiteral("sessionId"), session->sessionId); + } + const QByteArray bytes = QJsonDocument(request).toJson(QJsonDocument::Compact); + return service->handle(bytes, session).message; +} + +bool negotiate(ControlService* service, ControlSession* session) +{ + const QJsonObject reply = invoke( + service, session, QStringLiteral("hello"), QStringLiteral("hello"), + {{QStringLiteral("versions"), QJsonArray{1}}}); + return reply.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("sessionId")).toString() == session->sessionId + && session->negotiated; +} + +QString errorCode(const QJsonObject& response) +{ + return response.value(QStringLiteral("error")).toObject() + .value(QStringLiteral("code")).toString(); +} + +// The session hands the transport pre-encoded frames; decode them back so the +// assertions below can keep reading protocol fields by name. +QList drain(ControlSession& session) +{ + QList messages; + const QList frames = session.takePendingFrames(); + messages.reserve(frames.size()); + for (const QByteArray& frame : frames) { + messages.append(QJsonDocument::fromJson(frame).object()); + } + return messages; +} + +QJsonObject exactResource(const QString& type, const QString& radioSession = {}, + const QString& id = {}) +{ + QJsonObject resource{{QStringLiteral("type"), type}}; + if (!radioSession.isEmpty()) { + resource.insert(QStringLiteral("radioSession"), radioSession); + } + if (!id.isEmpty()) { + resource.insert(QStringLiteral("id"), id); + } + return resource; +} + +bool testStoreRevisions() +{ + ControlResourceStore store; + const ResourceAddress address{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + const ResourceAddress otherAddress{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("1")}; + quint64 removedRevision = 0; + QObject::connect(&store, &ControlResourceStore::resourceRemoved, + [&removedRevision](const ResourceAddress&, quint64 revision) { + removedRevision = revision; + }); + + if (!check(store.upsert(address, {{QStringLiteral("frequencyHz"), 100}}), + "first resource value must create revision 1") + || !check(!store.upsert(address, {{QStringLiteral("frequencyHz"), 100}}), + "an unchanged canonical value must not advance its revision") + || !check(store.upsert(otherAddress, {{QStringLiteral("frequencyHz"), 700}}), + "a second identity must consume the next store revision") + || !check(store.upsert(address, {{QStringLiteral("frequencyHz"), 200}}), + "a changed canonical value must advance its revision")) { + return false; + } + const std::optional second = store.get(address); + if (!check(second && second->revision == 3, + "store-wide revisions must stay monotonic across identities") + || !check(store.remove(address) && removedRevision == 4, + "removal must consume the next identity revision") + || !check(store.upsert(address, {{QStringLiteral("frequencyHz"), 300}}), + "a removed identity may be recreated")) { + return false; + } + const std::optional recreated = store.get(address); + return check(recreated && recreated->revision == 5, + "recreating an identity must not reset its revision"); +} + +bool testServiceSubscriptions() +{ + ControlResourceStore store; + store.upsert({QStringLiteral("server"), {}, {}}, + {{QStringLiteral("health"), QStringLiteral("ok")}}); + const ResourceAddress slice{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + store.upsert(slice, {{QStringLiteral("frequencyHz"), 100}}); + + ControlService service(&store); + ControlSession first(&store, 4096); + ControlSession second(&store, 4096); + if (!check(negotiate(&service, &first) && negotiate(&service, &second), + "both clients must negotiate independent sessions")) { + return false; + } + + const QJsonObject getReply = invoke( + &service, &first, QStringLiteral("get-1"), QStringLiteral("resource.get"), + {{QStringLiteral("resource"), + exactResource(QStringLiteral("slice"), QStringLiteral("radio-1"), + QStringLiteral("0"))}}); + const qint64 initialSliceRevision = getReply.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("revision")).toInteger(); + if (!check(initialSliceRevision > 0, + "resource.get must return the complete current snapshot")) { + return false; + } + + const QJsonArray sliceSelector{ + exactResource(QStringLiteral("slice"), QStringLiteral("radio-1"))}; + const QJsonObject firstSubscribe = invoke( + &service, &first, QStringLiteral("sub-1"), QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), sliceSelector}}); + const QJsonObject firstBaseline = firstSubscribe.value(QStringLiteral("result")).toObject(); + const QString firstSubscription = + firstBaseline.value(QStringLiteral("subscription")).toString(); + if (!check(firstBaseline.value(QStringLiteral("sequence")).toInteger() == 0 + && firstBaseline.value(QStringLiteral("resources")).toArray().size() == 1, + "subscribe must atomically return a sequence and matching baseline")) { + return false; + } + + store.upsert(slice, {{QStringLiteral("frequencyHz"), 200}}); + store.upsert(slice, {{QStringLiteral("frequencyHz"), 300}}); + const QList coalesced = drain(first); + if (!check(coalesced.size() == 1, + "undrained changes to one resource must coalesce") + || !check(coalesced.first().value(QStringLiteral("sequence")).toInteger() == 2 + && coalesced.first().value(QStringLiteral("revision")).toInteger() + == initialSliceRevision + 2 + && coalesced.first().value(QStringLiteral("value")).toObject() + .value(QStringLiteral("frequencyHz")).toInteger() == 300, + "the coalesced event must retain the newest sequence, revision, and value")) { + return false; + } + + const ResourceAddress secondSlice{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("1")}; + store.upsert(secondSlice, {{QStringLiteral("frequencyHz"), 700}}); + store.upsert(slice, {{QStringLiteral("frequencyHz"), 350}}); + const QList interleaved = drain(first); + if (!check(interleaved.size() == 2 + && interleaved.at(0).value(QStringLiteral("sequence")).toInteger() == 3 + && interleaved.at(1).value(QStringLiteral("sequence")).toInteger() == 4, + "coalescing must preserve event-sequence delivery order across resources")) { + return false; + } + + invoke(&service, &second, QStringLiteral("sub-2"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), sliceSelector}}); + store.upsert(slice, {{QStringLiteral("frequencyHz"), 400}}); + const QList firstNext = drain(first); + const QList secondNext = drain(second); + if (!check(firstNext.size() == 1 && secondNext.size() == 1 + && firstNext.first().value(QStringLiteral("sequence")).toInteger() == 5 + && secondNext.first().value(QStringLiteral("sequence")).toInteger() == 1, + "event sequences must be monotonic and local to each client session")) { + return false; + } + + store.upsert(slice, {{QStringLiteral("frequencyHz"), 450}}); + const QJsonObject unsubscribe = invoke( + &service, &first, QStringLiteral("unsub-1"), + QStringLiteral("resource.unsubscribe"), + {{QStringLiteral("subscription"), firstSubscription}}); + const QList afterUnsubscribe = drain(first); + store.upsert(slice, {{QStringLiteral("frequencyHz"), 500}}); + if (!check(unsubscribe.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("removed")).toBool() + && afterUnsubscribe.isEmpty() + && drain(first).isEmpty() + && drain(second).size() == 1, + "unsubscribe must discard undrained events no longer observed and isolate clients")) { + return false; + } + + const QJsonObject malformed = invoke( + &service, &first, QStringLiteral("bad-selector"), + QStringLiteral("resource.get"), + {{QStringLiteral("resource"), + QJsonObject{{QStringLiteral("type"), QStringLiteral("slice")}, + {QStringLiteral("radioSession"), QStringLiteral("radio-1")}, + {QStringLiteral("typo"), QStringLiteral("0")}}}}); + if (!check(errorCode(malformed) == QStringLiteral("request.invalid_params"), + "unknown selector fields must fail closed")) { + return false; + } + + const QJsonObject unknownSubscription = invoke( + &service, &first, QStringLiteral("unknown-unsub"), + QStringLiteral("resource.unsubscribe"), + {{QStringLiteral("subscription"), firstSubscription}}); + if (!check(errorCode(unknownSubscription) == QStringLiteral("resource.not_found"), + "unsubscribing an unknown ID must return resource.not_found")) { + return false; + } + + QJsonArray tooManySelectors; + for (int index = 0; index <= 64; ++index) { + tooManySelectors.append(sliceSelector.first()); + } + const QJsonObject selectorLimit = invoke( + &service, &first, QStringLiteral("selector-limit"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), tooManySelectors}}); + if (!check(errorCode(selectorLimit) == QStringLiteral("request.invalid_params"), + "a subscription must reject more than 64 selectors")) { + return false; + } + + ControlSession limited(&store, 4096); + if (!check(negotiate(&service, &limited), + "subscription-limit client must negotiate")) { + return false; + } + for (int index = 0; index < ControlSession::kMaxSubscriptions; ++index) { + const QJsonObject accepted = invoke( + &service, &limited, QStringLiteral("limit-%1").arg(index), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), sliceSelector}}); + if (!check(accepted.value(QStringLiteral("result")).isObject(), + "the declared number of subscriptions must be accepted")) { + return false; + } + } + const QJsonObject subscriptionLimit = invoke( + &service, &limited, QStringLiteral("limit-overflow"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), sliceSelector}}); + return check(errorCode(subscriptionLimit) + == QStringLiteral("transport.limit_exceeded"), + "a client must reject subscription 65"); +} + +bool testSubscribeSequenceBoundary() +{ + ControlResourceStore store; + ControlService service(&store); + ControlSession session(&store, 4096); + if (!check(negotiate(&service, &session), + "sequence-boundary client must negotiate")) { + return false; + } + + const ResourceAddress slice{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + store.upsert(slice, {{QStringLiteral("frequencyHz"), 100}}); + const QJsonArray selectors{ + exactResource(QStringLiteral("slice"), QStringLiteral("radio-1"))}; + invoke(&service, &session, QStringLiteral("sub-initial"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}); + + store.upsert(slice, {{QStringLiteral("frequencyHz"), 200}}); + const QJsonObject pendingBaseline = invoke( + &service, &session, QStringLiteral("sub-pending"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}) + .value(QStringLiteral("result")).toObject(); + const QList pending = drain(session); + if (!check(pendingBaseline.value(QStringLiteral("sequence")).toInteger() == 0 + && pending.size() == 1 + && pending.first().value(QStringLiteral("sequence")).toInteger() == 1, + "a baseline must stop at the last drained sequence when an older event is pending")) { + return false; + } + + const QJsonObject drainedBaseline = invoke( + &service, &session, QStringLiteral("sub-drained"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}) + .value(QStringLiteral("result")).toObject(); + store.upsert(slice, {{QStringLiteral("frequencyHz"), 300}}); + const QList afterBaseline = drain(session); + return check(drainedBaseline.value(QStringLiteral("sequence")).toInteger() == 1 + && afterBaseline.size() == 1 + && afterBaseline.first().value(QStringLiteral("sequence")).toInteger() == 2, + "the next delivered event must have a sequence greater than the baseline"); +} + +bool testOverflowRequiresResync() +{ + ControlResourceStore store; + ControlService service(&store); + ControlSession session(&store, 360); + if (!check(negotiate(&service, &session), "overflow client must negotiate")) { + return false; + } + const QJsonArray selectors{ + exactResource(QStringLiteral("slice"), QStringLiteral("radio-1"))}; + invoke(&service, &session, QStringLiteral("sub"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}); + + store.upsert({QStringLiteral("slice"), QStringLiteral("radio-1"), QStringLiteral("0")}, + {{QStringLiteral("value"), QString(100, QLatin1Char('a'))}}); + store.upsert({QStringLiteral("slice"), QStringLiteral("radio-1"), QStringLiteral("1")}, + {{QStringLiteral("value"), QString(100, QLatin1Char('b'))}}); + const QList overflow = drain(session); + if (!check(overflow.size() == 1 + && overflow.first().value(QStringLiteral("event")).toString() + == QStringLiteral("resource.resyncRequired") + && overflow.first().value(QStringLiteral("subscriptionsInvalidated")).toBool(), + "queue overflow must invalidate subscriptions and require an explicit resync")) { + return false; + } + + store.upsert({QStringLiteral("slice"), QStringLiteral("radio-1"), QStringLiteral("0")}, + {{QStringLiteral("value"), QStringLiteral("after-overflow")}}); + if (!check(drain(session).isEmpty(), + "invalidated subscriptions must remain quiet until resubscribed")) { + return false; + } + const QJsonObject resubscribe = invoke( + &service, &session, QStringLiteral("resub"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}); + if (!check(resubscribe.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("resources")).toArray().size() == 2, + "resubscribe must establish a fresh complete baseline")) { + return false; + } + + store.upsert({QStringLiteral("slice"), QStringLiteral("radio-1"), QStringLiteral("0")}, + {{QStringLiteral("value"), QString(100, QLatin1Char('c'))}}); + store.upsert({QStringLiteral("slice"), QStringLiteral("radio-1"), QStringLiteral("1")}, + {{QStringLiteral("value"), QString(100, QLatin1Char('d'))}}); + const QJsonObject baselineAfterUndrainedResync = invoke( + &service, &session, QStringLiteral("resub-undrained"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}); + return check(baselineAfterUndrainedResync.value(QStringLiteral("result")).isObject() + && drain(session).isEmpty(), + "a fresh baseline must supersede an undrained resync notice"); +} + +RadioInfo demoInfo() +{ + RadioInfo info; + info.name = SimBackend::demoModelName(); + info.model = SimBackend::demoModelName(); + info.serial = SimBackend::demoSerial(); + info.family = SimBackend::familyName(); + info.address = QHostAddress(QHostAddress::LocalHost); + info.port = 4992; + return info; +} + +bool testWireFramingIsCanonical() +{ + // ControlSession now encodes each event once and the transport writes the + // stored frame verbatim, so the frame it hands over must still be exactly + // what LocalControlServer::send() would have produced: compact JSON plus + // the single framing newline, and nothing else. + ControlResourceStore store; + ControlService service(&store); + ControlSession session(&store, 4096); + if (!check(negotiate(&service, &session), "framing client must negotiate")) { + return false; + } + const ResourceAddress slice{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + store.upsert(slice, {{QStringLiteral("frequencyHz"), 100}}); + invoke(&service, &session, QStringLiteral("frame-sub"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), + QJsonArray{exactResource(QStringLiteral("slice"), + QStringLiteral("radio-1"))}}}); + store.upsert(slice, {{QStringLiteral("frequencyHz"), 200}}); + + const QList frames = session.takePendingFrames(); + if (!check(frames.size() == 1, "one change must produce one frame")) { + return false; + } + const QByteArray frame = frames.first(); + QJsonParseError parseError{}; + const QJsonDocument document = QJsonDocument::fromJson(frame, &parseError); + const QJsonObject message = document.object(); + return check(frame.endsWith('\n') && !frame.chopped(1).contains('\n'), + "a frame must be one line terminated by a single newline") + && check(parseError.error == QJsonParseError::NoError && document.isObject(), + "a frame must decode as one JSON object") + && check(QJsonDocument(message).toJson(QJsonDocument::Compact) + '\n' == frame, + "a frame must be compact JSON, byte-identical to the transport encoding") + && check(message.value(QStringLiteral("event")).toString() + == QStringLiteral("resource.changed") + && message.value(QStringLiteral("sequence")).toInteger() == 1 + && message.value(QStringLiteral("sessionId")).toString() + == session.sessionId, + "a frame must carry the session-sequenced event envelope"); +} + +bool testPureSeamReconnectRepublishesSlice() +{ + ControlResourceStore store; + RadioModel radio; + RadioResourceAdapter adapter(&radio, &store, QStringLiteral("radio-1")); + radio.connectToRadio(demoInfo()); + + const ResourceAddress sliceAddress{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + if (!check(waitUntil([&] { return store.get(sliceAddress).has_value(); }), + "the seam-reconnect fixture must start with a published slice")) { + radio.disconnectFromRadio(); + return false; + } + SliceModel* originalSlice = radio.slice(0); + if (!check(originalSlice != nullptr, + "the seam-reconnect fixture must retain its SliceModel")) { + radio.disconnectFromRadio(); + return false; + } + + radio.disconnectFromRadio(); + if (!check(waitUntil([&] { + return !radio.isConnected() + && !store.get(sliceAddress).has_value(); + }), + "disconnect must remove the published slice before reclaim")) { + return false; + } + + radio.stageSessionModelsForReconnectForTest(); + radio.connectionStateChanged(true); + int occupancySignals = 0; + QObject::connect(&radio, &RadioModel::slotOccupancyChanged, + [&occupancySignals](int sliceId) { + if (sliceId == 0) { + ++occupancySignals; + } + }); + SliceDelta delta; + delta.frequency = originalSlice->frequency(); + delta.mode = originalSlice->mode(); + delta.panId = originalSlice->panId(); + radio.emitBackendSliceChangedForTest(0, delta); + + const std::optional reclaimed = store.get(sliceAddress); + const bool result = check(radio.slice(0) == originalSlice, + "the normalized backend seam must reclaim the existing SliceModel") + && check(occupancySignals == 1, + "non-Flex slice reclaim must publish an occupancy edge") + && check(reclaimed.has_value() + && reclaimed->value.value(QStringLiteral("owned")).toBool(), + "the occupancy edge must reattach and republish the slice resource"); + radio.connectionStateChanged(false); + return result; +} + +bool testExternalReceiveAudioRepublishesAgcAndSquelch() +{ + ControlResourceStore store; + RadioModel radio; + RadioResourceAdapter adapter(&radio, &store, QStringLiteral("radio-1")); + radio.connectToRadio(demoInfo()); + + const ResourceAddress sliceAddress{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + if (!check(waitUntil([&] { return store.get(sliceAddress).has_value(); }), + "the external-receive fixture must start with a published slice")) { + radio.disconnectFromRadio(); + return false; + } + SliceModel* slice = radio.slice(0); + if (!check(slice != nullptr, "the external-receive fixture must retain its SliceModel")) { + radio.disconnectFromRadio(); + return false; + } + + // While an external receive-audio source replaces the radio's audio, the + // AGC and squelch setters emit only their externalReceive* signals but the + // accessors the adapter reads switch to the external values. + slice->setExternalReceiveAudioReplacementMute(true); + slice->setAgcMode(QStringLiteral("fast")); + slice->setAgcThreshold(-40); + slice->setSquelch(true, 42); + + const auto receiveOf = [&store, &sliceAddress] { + return store.get(sliceAddress)->value.value(QStringLiteral("receive")).toObject(); + }; + const QJsonObject receive = receiveOf(); + const QJsonObject agc = receive.value(QStringLiteral("agc")).toObject(); + const QJsonObject squelch = receive.value(QStringLiteral("squelch")).toObject(); + const bool result = + check(agc.value(QStringLiteral("mode")).toString() == slice->receiveAgcMode() + && agc.value(QStringLiteral("mode")).toString() + == QStringLiteral("fast"), + "external receive AGC mode must reach the published resource") + && check(agc.value(QStringLiteral("threshold")).toInt() + == slice->receiveAgcThreshold(), + "external receive AGC threshold must reach the published resource") + && check(squelch.value(QStringLiteral("enabled")).toBool() + == slice->receiveSquelchOn() + && squelch.value(QStringLiteral("level")).toInt() + == slice->receiveSquelchLevel() + && squelch.value(QStringLiteral("level")).toInt() == 42, + "external receive squelch must reach the published resource"); + radio.disconnectFromRadio(); + waitUntil([&radio] { return !radio.isConnected(); }); + return result; +} + +bool testSliceRemovalMatchesTheLiveObject() +{ + ControlResourceStore store; + RadioModel radio; + RadioResourceAdapter adapter(&radio, &store, QStringLiteral("radio-1")); + radio.connectToRadio(demoInfo()); + + const ResourceAddress sliceAddress{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + if (!check(waitUntil([&] { return store.get(sliceAddress).has_value(); }), + "the slice-removal fixture must start with a published slice")) { + radio.disconnectFromRadio(); + return false; + } + + // pruneStaleSessionModels() emits sliceRemoved for a *stale* SliceModel whose + // id a live one may already have reclaimed. An id-only match would unbind the + // live slice and publish a removal for a resource that still exists. + radio.sliceRemoved(0); + if (!check(store.get(sliceAddress).has_value(), + "a removal for an id that still resolves to a live slice must keep it")) { + radio.disconnectFromRadio(); + return false; + } + const quint64 revision = store.get(sliceAddress)->revision; + SliceModel* slice = radio.slice(0); + if (!check(slice != nullptr, "the live slice must survive the stale removal")) { + radio.disconnectFromRadio(); + return false; + } + slice->setLocked(!slice->isLocked()); + if (!check(store.get(sliceAddress) + && store.get(sliceAddress)->revision > revision, + "the live slice must stay attached after a stale removal")) { + radio.disconnectFromRadio(); + return false; + } + + // Once the model no longer serves the id, the same signal must drop it. + radio.stageSessionModelsForReconnectForTest(); + radio.sliceRemoved(0); + const bool result = check(!store.get(sliceAddress).has_value(), + "a removal for an id the model no longer serves must drop it"); + radio.disconnectFromRadio(); + waitUntil([&radio] { return !radio.isConnected(); }); + return result; +} + +bool testSimBackendEndToEnd() +{ + ControlResourceStore store; + ControlService service(&store); + ControlSession client(&store, 1024 * 1024); + RadioModel radio; + RadioResourceAdapter adapter(&radio, &store, QStringLiteral("radio-1")); + if (!check(negotiate(&service, &client), "sim observer must negotiate")) { + return false; + } + + const QJsonArray selectors{ + exactResource(QStringLiteral("radioSession")), + exactResource(QStringLiteral("slice"), QStringLiteral("radio-1")), + exactResource(QStringLiteral("panadapter"), QStringLiteral("radio-1"))}; + const QJsonObject baselineReply = invoke( + &service, &client, QStringLiteral("sim-sub"), + QStringLiteral("resource.subscribe"), + {{QStringLiteral("resources"), selectors}}); + if (!check(baselineReply.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("resources")).toArray().size() == 1, + "the disconnected model must begin as one radioSession resource")) { + return false; + } + + radio.connectToRadio(demoInfo()); + const ResourceAddress radioAddress{QStringLiteral("radioSession"), {}, + QStringLiteral("radio-1")}; + const ResourceAddress sliceAddress{QStringLiteral("slice"), + QStringLiteral("radio-1"), QStringLiteral("0")}; + const ResourceAddress panAddress{QStringLiteral("panadapter"), + QStringLiteral("radio-1"), + QStringLiteral("0x40000000")}; + const bool converged = waitUntil([&] { + const std::optional radioSnapshot = store.get(radioAddress); + return radioSnapshot + && radioSnapshot->value.value(QStringLiteral("connected")).toBool() + && store.get(sliceAddress).has_value() + && store.get(panAddress).has_value(); + }); + if (!check(converged, + "SimBackend must drive connected radio, slice, and panadapter resources")) { + radio.disconnectFromRadio(); + return false; + } + + const QJsonObject sliceReply = invoke( + &service, &client, QStringLiteral("sim-get"), + QStringLiteral("resource.get"), + {{QStringLiteral("resource"), + exactResource(QStringLiteral("slice"), QStringLiteral("radio-1"), + QStringLiteral("0"))}}); + const QJsonObject sliceValue = sliceReply.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("value")).toObject(); + const QJsonObject radioValue = store.get(radioAddress)->value; + const QJsonObject panValue = store.get(panAddress)->value; + const QJsonObject identity = radioValue.value(QStringLiteral("identity")).toObject(); + const QJsonObject capabilities = + radioValue.value(QStringLiteral("capabilities")).toObject(); + const QJsonObject receive = sliceValue.value(QStringLiteral("receive")).toObject(); + const QJsonObject displayCadence = + panValue.value(QStringLiteral("displayCadence")).toObject(); + if (!check(hasExactlyKeys(radioValue, + {"id", "connected", "family", "identity", "capabilities"}) + && hasExactlyKeys(identity, + {"name", "model", "serial", "version", "manufacturer"}) + && hasExactlyKeys(capabilities, + {"maxSlices", "maxPanadapters", "sampleRatesHz", + "tuningRangeHz", "declaredBands", "canTransmit", + "maximumTransmitWatts", "hasTuner", "hasAmplifier", + "extensions"}) + && hasExactlyKeys(sliceValue, + {"id", "letter", "panadapterId", "owned", + "frequencyHz", "mode", "filter", "active", + "txSlice", "locked", "audio", "receive"}) + && hasExactlyKeys(sliceValue.value(QStringLiteral("filter")).toObject(), + {"lowHz", "highHz"}) + && hasExactlyKeys(sliceValue.value(QStringLiteral("audio")).toObject(), + {"gain", "pan", "muted"}) + && hasExactlyKeys(receive, + {"antenna", "rfGain", "agc", "squelch"}) + && hasExactlyKeys(receive.value(QStringLiteral("agc")).toObject(), + {"mode", "threshold", "offLevel"}) + && hasExactlyKeys(receive.value(QStringLiteral("squelch")).toObject(), + {"enabled", "level"}) + && hasExactlyKeys(panValue, + {"id", "centerHz", "centerKnown", + "bandwidthHz", "dbmRange", "bandwidthLimitsHz", + "receive", "displayCadence"}) + && hasExactlyKeys(panValue.value(QStringLiteral("dbmRange")).toObject(), + {"minimum", "maximum"}) + && hasExactlyKeys( + panValue.value(QStringLiteral("bandwidthLimitsHz")).toObject(), + {"minimum", "maximum"}) + && hasExactlyKeys(panValue.value(QStringLiteral("receive")).toObject(), + {"antenna", "rfGain"}) + && hasExactlyKeys(displayCadence, + {"fps", "averageFrames", "weightedAverage", + "weightedAverageKnown", "waterfallRate"}), + "SimBackend resources must match the complete documented v1 schemas")) { + radio.disconnectFromRadio(); + return false; + } + + QList events = drain(client); + PanadapterModel* modelPan = radio.panadapter(QStringLiteral("0x40000000")); + if (!check(modelPan && !modelPan->weightedAverageKnown(), + "weighted averaging must begin unknown before its first report")) { + radio.disconnectFromRadio(); + return false; + } + const quint64 weightedRevision = store.get(panAddress)->revision; + modelPan->applyStateExtension( + {{QStringLiteral("weighted_average"), QStringLiteral("0")}}); + const std::optional reportedWeighted = store.get(panAddress); + const QList weightedEvents = drain(client); + events.append(weightedEvents); + if (!check(reportedWeighted && reportedWeighted->revision > weightedRevision + && reportedWeighted->value.value(QStringLiteral("displayCadence")) + .toObject().value(QStringLiteral("weightedAverageKnown")).toBool() + && !reportedWeighted->value.value(QStringLiteral("displayCadence")) + .toObject().value(QStringLiteral("weightedAverage")).toBool() + && weightedEvents.size() == 1 + && weightedEvents.first().value(QStringLiteral("event")).toString() + == QStringLiteral("resource.changed"), + "the first known-false weighted-average report must publish an event")) { + radio.disconnectFromRadio(); + return false; + } + const quint64 knownWeightedRevision = reportedWeighted->revision; + modelPan->applyStateExtension( + {{QStringLiteral("weighted_average"), QStringLiteral("0")}}); + if (!check(store.get(panAddress)->revision == knownWeightedRevision + && drain(client).isEmpty(), + "an identical known weighted-average report must be deduplicated")) { + radio.disconnectFromRadio(); + return false; + } + + const quint64 sliceRevision = store.get(sliceAddress)->revision; + QJsonObject staleSliceValue = sliceValue; + staleSliceValue.insert(QStringLiteral("owned"), false); + store.upsert(sliceAddress, staleSliceValue); + radio.slotOccupancyChanged(0); + const std::optional refreshedSlice = store.get(sliceAddress); + if (!check(refreshedSlice && refreshedSlice->revision > sliceRevision + && refreshedSlice->value.value(QStringLiteral("owned")).toBool(), + "slot occupancy changes must republish RadioModel-derived ownership")) { + radio.disconnectFromRadio(); + return false; + } + + const quint64 panRevision = store.get(panAddress)->revision; + QJsonObject stalePanValue = panValue; + stalePanValue.insert(QStringLiteral("centerKnown"), + !panValue.value(QStringLiteral("centerKnown")).toBool()); + store.upsert(panAddress, stalePanValue); + radio.panadapterReclaimed(radio.panadapter(QStringLiteral("0x40000000"))); + const std::optional refreshedPan = store.get(panAddress); + if (!check(refreshedPan && refreshedPan->revision > panRevision + && refreshedPan->value.value(QStringLiteral("centerKnown")) + == panValue.value(QStringLiteral("centerKnown")), + "panadapter reclaim must republish canonical model state")) { + radio.disconnectFromRadio(); + return false; + } + events.append(drain(client)); + bool sawRadio = false; + bool sawSlice = false; + bool sawPan = false; + qint64 previousSequence = 0; + for (const QJsonObject& event : events) { + const qint64 sequence = event.value(QStringLiteral("sequence")).toInteger(); + if (sequence <= previousSequence) { + radio.disconnectFromRadio(); + return check(false, "SimBackend events must remain ordered after coalescing"); + } + previousSequence = sequence; + const QString type = event.value(QStringLiteral("resource")).toObject() + .value(QStringLiteral("type")).toString(); + sawRadio = sawRadio || type == QStringLiteral("radioSession"); + sawSlice = sawSlice || type == QStringLiteral("slice"); + sawPan = sawPan || type == QStringLiteral("panadapter"); + } + + const SliceModel* modelSlice = radio.slice(0); + const bool result = + check(modelSlice + && sliceValue.value(QStringLiteral("frequencyHz")).toInteger() + == qRound64(modelSlice->frequency() * 1'000'000.0) + && sliceValue.value(QStringLiteral("mode")).toString() + == modelSlice->mode(), + "resource.get must expose SimBackend's authoritative slice state") + && check(sawRadio && sawSlice && sawPan, + "the subscription must deliver each SimBackend resource family") + && check(!radio.backendCapabilities().canTransmit, + "the end-to-end simulator proof must remain RX-only"); + radio.disconnectFromRadio(); + const bool removed = waitUntil([&] { + return !radio.isConnected() + && !store.get(sliceAddress).has_value() + && !store.get(panAddress).has_value(); + }); + if (!result + || !check(removed, + "disconnect must remove dynamic resources and update the radio session")) { + return false; + } + + radio.connectToRadio(demoInfo()); + const bool reclaimed = waitUntil([&] { + return radio.isConnected() + && store.get(sliceAddress).has_value() + && store.get(panAddress).has_value(); + }); + radio.disconnectFromRadio(); + waitUntil([&radio] { return !radio.isConnected(); }); + return check(reclaimed, + "reconnect must republish reclaimed slice and panadapter resources"); +} + +} // namespace + +int main(int argc, char* argv[]) +{ + TestSettingsProfile settingsProfile( + QStringLiteral("aether-control-resource-service-test")); + if (!check(settingsProfile.isValid(), + "isolated settings profile must be available")) { + return 1; + } + QCoreApplication app(argc, argv); + return testStoreRevisions() + && testServiceSubscriptions() + && testSubscribeSequenceBoundary() + && testOverflowRequiresResync() + && testWireFramingIsCanonical() + && testPureSeamReconnectRepublishesSlice() + && testExternalReceiveAudioRepublishesAgcAndSquelch() + && testSliceRemovalMatchesTheLiveObject() + && testSimBackendEndToEnd() ? 0 : 1; +} diff --git a/tests/local_control_server_test.cpp b/tests/local_control_server_test.cpp index f9f7796be..193cda15a 100644 --- a/tests/local_control_server_test.cpp +++ b/tests/local_control_server_test.cpp @@ -21,6 +21,7 @@ #include using AetherSDR::control::LocalControlServer; +using AetherSDR::control::ControlSession; using AetherSDR::control::ProtocolLimits; namespace { @@ -152,16 +153,27 @@ bool runProtocolTest() if (!check(!sessionId.isEmpty(), "hello must create a sessionId") || !check(grants.size() == 1 && contains(grants, QStringLiteral("observe")), "server must grant observe only") - || !check(capabilities.size() == 1 - && contains(capabilities, QStringLiteral("server.read")), - "server must advertise server.read only") + || !check(capabilities.size() == 7 + && contains(capabilities, QStringLiteral("server.read")) + && contains(capabilities, QStringLiteral("radioSession.read")) + && contains(capabilities, QStringLiteral("slice.read")) + && contains(capabilities, QStringLiteral("panadapter.read")) + && contains(capabilities, QStringLiteral("resource.get")) + && contains(capabilities, QStringLiteral("resource.subscribe")) + && contains(capabilities, QStringLiteral("resource.unsubscribe")), + "server must advertise the implemented read-resource surface") || !check(!contains(capabilities, QStringLiteral("transmit")) && !contains(capabilities, QStringLiteral("control")), "server must not advertise control or transmit") - || !check(limits.size() == 1 - && limits.value(QStringLiteral("maxMessageBytes")).toInteger() - == ProtocolLimits::kMaxMessageBytes, - "server must advertise only its currently enforced protocol limit")) { + || !check(limits.value(QStringLiteral("maxMessageBytes")).toInteger() + == ProtocolLimits::kMaxMessageBytes + && limits.value(QStringLiteral("maxSubscriptions")).toInteger() + == ControlSession::kMaxSubscriptions + && limits.value(QStringLiteral("maxSelectorsPerSubscription")).toInteger() + == 64 + && limits.value(QStringLiteral("maxQueuedOutputBytes")).toInteger() + == LocalControlServer::kMaxQueuedOutputBytes, + "server must advertise its enforced resource limits")) { return false; } @@ -177,6 +189,54 @@ bool runProtocolTest() return false; } + const QJsonObject serverResource{{QStringLiteral("type"), + QStringLiteral("server")}}; + const QJsonObject resourceReply = exchange(&socket, { + {QStringLiteral("v"), 1}, + {QStringLiteral("id"), QStringLiteral("resource-1")}, + {QStringLiteral("sessionId"), sessionId}, + {QStringLiteral("method"), QStringLiteral("resource.get")}, + {QStringLiteral("params"), QJsonObject{ + {QStringLiteral("resource"), serverResource}}} + }); + if (!check(resourceReply.value(QStringLiteral("result")).toObject() + .value(QStringLiteral("value")).toObject() + .value(QStringLiteral("localTransport")).toString() + == QStringLiteral("listening"), + "resource.get must cross the local transport")) { + return false; + } + + const QJsonObject subscribeReply = exchange(&socket, { + {QStringLiteral("v"), 1}, + {QStringLiteral("id"), QStringLiteral("subscribe-1")}, + {QStringLiteral("sessionId"), sessionId}, + {QStringLiteral("method"), QStringLiteral("resource.subscribe")}, + {QStringLiteral("params"), QJsonObject{ + {QStringLiteral("resources"), QJsonArray{serverResource}}}} + }); + const QJsonObject subscribeResult = + subscribeReply.value(QStringLiteral("result")).toObject(); + if (!check(subscribeResult.value(QStringLiteral("resources")).toArray().size() == 1, + "resource.subscribe must return its atomic baseline")) { + return false; + } + const AetherSDR::control::ResourceAddress serverAddress{ + QStringLiteral("server"), {}, {}}; + QJsonObject updatedServer = server.resourceStore().get(serverAddress)->value; + updatedServer.insert(QStringLiteral("health"), QStringLiteral("test-update")); + server.resourceStore().upsert(serverAddress, updatedServer); + const QJsonObject event = readResponse(&socket); + if (!check(event.value(QStringLiteral("event")).toString() + == QStringLiteral("resource.changed") + && event.value(QStringLiteral("sequence")).toInteger() == 1 + && event.value(QStringLiteral("value")).toObject() + .value(QStringLiteral("health")).toString() + == QStringLiteral("test-update"), + "resource changes must be delivered asynchronously over the local transport")) { + return false; + } + const QJsonObject wrongSession = exchange(&socket, { {QStringLiteral("v"), 1}, {QStringLiteral("id"), QStringLiteral("caps-2")}, diff --git a/tests/tests.cmake b/tests/tests.cmake index d50f36aa2..229cd5d41 100644 --- a/tests/tests.cmake +++ b/tests/tests.cmake @@ -88,6 +88,19 @@ target_link_libraries(local_control_server_test PRIVATE aethercore Qt6::Core Qt6::Network) add_test(NAME local_control_server_test COMMAND local_control_server_test) +# Socket-free Stage-3 resource/service proof: revision stability, atomic +# snapshot-to-event sequencing, multi-client delivery, unsubscribe, +# coalescing/resync under pressure, normalized backend reconnect reclaim, +# external receive-audio AGC/squelch republish, stale-vs-live slice removal, and +# SimBackend -> RadioModel -> protocol. +add_executable(control_resource_service_test + tests/control_resource_service_test.cpp +) +target_include_directories(control_resource_service_test PRIVATE src) +target_link_libraries(control_resource_service_test PRIVATE + aethercore Qt6::Core) +add_test(NAME control_resource_service_test COMMAND control_resource_service_test) + # ── Digital-voice / D-STAR tests ───────────────────────────────────────────── # Guarded by the same condition as the aether-dv-waveform target they exercise. # DIGITAL_VOICE_WAVEFORM_DIR, CRDV_DIR and crdv::crdv are all defined by the time @@ -4300,6 +4313,7 @@ target_link_libraries(CAT_Flex_test PRIVATE Qt6::Core Qt6::Network) # directly (rather than linking aethercore) needs the vendored SQLite engine. # Conditional targets are guarded with if(TARGET ...). set(AETHER_SETTINGS_CONSUMERS + control_resource_service_test slice_label_test ulanzi_mapping_migration_test theme_manager_test