From 50f10eb341fceff7a0b79a057df1472017afe811 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Tue, 9 Jun 2026 14:30:58 +0200 Subject: [PATCH 1/4] py: add get() --- CHANGELOG.md | 2 -- py/README.md | 2 ++ py/miniconf/cli.py | 6 +----- py/miniconf/client.py | 14 ++++++++++++++ 4 files changed, 17 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 46343412..cfcceaae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,8 +33,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * `miniconf_mqtt` MM2 now publishes retained manifest, paged schema, and authoritative retained settings. Compatibility with request/response-only settings writes is retained behind the `compat-settings-ingress` feature. -* MM2 manifests now publish `epoch` and `schema_rev`. Long-lived Python clients use them to - invalidate cached schema and settings. * The Python package was restructured from `py/miniconf-mqtt` to `py/` and now targets the MM2 retained schema/settings protocol with an async-first CLI and client library. diff --git a/py/README.md b/py/README.md index 3755a04d..feef0276 100644 --- a/py/README.md +++ b/py/README.md @@ -18,6 +18,7 @@ from miniconf.common import MQTTv5 async with Client("mqtt", protocol=MQTTv5) as mqtt: async with Miniconf(mqtt, "app/id") as mc: schema = await mc.schema() + value = await mc.get("/path") await mc.set("/path", 42) async with mc.track("/subtree") as tracked: @@ -29,6 +30,7 @@ async with Client("mqtt", protocol=MQTTv5) as mqtt: Core API: - `schema()` loads and caches the retained schema. +- `get(path)` reads one schema-validated retained leaf without opening a subtree cache. - `set(path, value, response=True)` publishes one `set/#` request. - `track(path="")` scopes a retained subtree cache; `.ready` reports whether it is usable, `wait_ready()` waits through reboot/reload, then `cached()` reads one leaf diff --git a/py/miniconf/cli.py b/py/miniconf/cli.py index 8925636d..f6f3ecf0 100644 --- a/py/miniconf/cli.py +++ b/py/miniconf/cli.py @@ -227,11 +227,7 @@ def normalize(path: str) -> str: print(f"{path}={value}") else: path, base = _normalize_command_path(arg, base, subtree=False) - if raw: - value = await interface.get(path, timeout=timeout) - else: - async with interface.track(path, timeout=timeout) as tracked: - value = tracked.cached() + value = await interface.get(path, timeout=timeout) print(f"{path}={json_dumps(value)}") except (MiniconfException, TimeoutError, json.JSONDecodeError) as err: print(f"{arg}: {err!r}") diff --git a/py/miniconf/client.py b/py/miniconf/client.py index 7401e89a..84e157f3 100644 --- a/py/miniconf/client.py +++ b/py/miniconf/client.py @@ -477,6 +477,20 @@ async def set( path, json_dumps(value), response=response, timeout=timeout ) + async def get(self, path: str, *, timeout: float = 3.0): + """Read one schema-validated retained authoritative leaf.""" + + schema = await self.schema(timeout=timeout) + path = schema.path(path) + if schema.node(path).kind != "leaf": + raise MiniconfException("LeafRequired", path) + return await _read_retained_json( + self._watch, + f"{self.prefix}/settings{path}", + path, + timeout=timeout, + ) + async def schema(self, *, timeout: float = 3.0) -> Schema: """Load and cache the retained paged schema.""" From d6a4659d0c476df993f8ce3977ee844dc460708b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Tue, 9 Jun 2026 20:38:34 +0200 Subject: [PATCH 2/4] py: gmqtt and streamline client like web --- CHANGELOG.md | 2 + py/README.md | 39 ++- py/miniconf/_mqtt.py | 220 +++++++++++++++++ py/miniconf/_ops.py | 71 +++--- py/miniconf/cli.py | 19 +- py/miniconf/client.py | 560 ++++++++++++++++-------------------------- py/miniconf/common.py | 51 ++-- py/pyproject.toml | 2 +- py/test.py | 144 +++++++---- 9 files changed, 616 insertions(+), 492 deletions(-) create mode 100644 py/miniconf/_mqtt.py diff --git a/CHANGELOG.md b/CHANGELOG.md index cfcceaae..956209f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * `miniconf_mqtt` MM2 now publishes retained manifest, paged schema, and authoritative retained settings. Compatibility with request/response-only settings writes is retained behind the `compat-settings-ingress` feature. +* MM2 manifests now publish `epoch` and `schema_rev`. Long-lived Python clients use `schema_rev` + to invalidate cached schema. * The Python package was restructured from `py/miniconf-mqtt` to `py/` and now targets the MM2 retained schema/settings protocol with an async-first CLI and client library. diff --git a/py/README.md b/py/README.md index feef0276..40cc6655 100644 --- a/py/README.md +++ b/py/README.md @@ -11,20 +11,16 @@ python -m pip install -e py/ Use the async client for schema-aware access: ```python -from aiomqtt import Client from miniconf.client import Miniconf -from miniconf.common import MQTTv5 - -async with Client("mqtt", protocol=MQTTv5) as mqtt: - async with Miniconf(mqtt, "app/id") as mc: - schema = await mc.schema() - value = await mc.get("/path") - await mc.set("/path", 42) - - async with mc.track("/subtree") as tracked: - await tracked.wait_ready() - value = tracked.cached("/subtree/leaf") - cached = tracked.snapshot() + +async with Miniconf.connect("mqtt", "app/id") as mc: + schema = await mc.schema() + value = await mc.get("/path") + snapshot = await mc.snapshot("/subtree") + await mc.set("/path", 42) + + async for event in mc.watch("/subtree"): + print(event.path, event.value if event.present else "") ``` Core API: @@ -32,10 +28,11 @@ Core API: - `schema()` loads and caches the retained schema. - `get(path)` reads one schema-validated retained leaf without opening a subtree cache. - `set(path, value, response=True)` publishes one `set/#` request. -- `track(path="")` scopes a retained subtree cache; `.ready` reports whether it is - usable, `wait_ready()` waits through reboot/reload, then `cached()` reads one leaf - and `snapshot()` reads a subtree. -- `RawMiniconf` provides exact-path `get()` and `set()` without schema loading. +- `snapshot(path="")` reads a finite retained subtree snapshot. +- `watch(path="")` streams authoritative retained settings publications below a subtree without + waiting for quiescence. Events distinguish JSON `null` from retained deletes through `.present`. +- `RawMiniconf` provides exact-path `get()`, `set()`, `snapshot()`, and `watch()` without schema + loading. Schema helpers: @@ -48,9 +45,11 @@ Schema helpers: Notes: -- The client accepts the MM2 wire protocol `proto=1`, keeps `/alive` subscribed, and - reloads tracked settings when `epoch` changes; it reloads schema when `schema_rev` changes. -- Schema-aware reads are explicit: open `track()` for the subtree you want, then read its cache. +- The client accepts the MM2 wire protocol `proto=1`, keeps `/alive` subscribed, and reloads schema + when `schema_rev` changes. +- Exact reads and open watches do not wait for subtree quiescence. +- Finite retained subtree snapshots use a quiescence window because MQTT retained replay has no + end-of-set marker. - Retained `/settings` messages without `auth=""` are ignored as non-authoritative settings traffic. - Retained burst quiescence uses the same rule as the Rust client: diff --git a/py/miniconf/_mqtt.py b/py/miniconf/_mqtt.py new file mode 100644 index 00000000..51bbd597 --- /dev/null +++ b/py/miniconf/_mqtt.py @@ -0,0 +1,220 @@ +"""Small gmqtt transport used by the Miniconf protocol client.""" + +from __future__ import annotations + +import asyncio +import uuid +from dataclasses import dataclass +from typing import Any + +from gmqtt import Client as GmqttClient +from gmqtt import Subscription as GmqttSubscription +from gmqtt.mqtt.constants import MQTTv50 +from gmqtt.mqtt.handler import MQTTError + +MQTTv5 = MQTTv50 + + +@dataclass(frozen=True) +class Message: + topic: str + payload: bytes + retain: bool + properties: dict[str, Any] + + +class _Messages: + def __init__(self): + self._queue: asyncio.Queue[Message] = asyncio.Queue() + + def __aiter__(self): + return self + + async def __anext__(self) -> Message: + return await self._queue.get() + + def put(self, message: Message) -> None: + self._queue.put_nowait(message) + + +def _first(properties: dict[str, Any], name: str) -> Any: + value = properties.get(name) + if isinstance(value, list): + return value[0] if value else None + return value + + +def _normalize_properties(properties: dict[str, Any]) -> dict[str, Any]: + normalized: dict[str, Any] = {} + if payload_format := _first(properties, "payload_format_id"): + normalized["payload_format_id"] = payload_format + if expiry := _first(properties, "message_expiry_interval"): + normalized["message_expiry_interval"] = expiry + if response_topic := _first(properties, "response_topic"): + normalized["response_topic"] = response_topic + correlation = _first(properties, "correlation_data") + if correlation is not None: + normalized["correlation_data"] = correlation + user_properties = properties.get("user_property") + if user_properties is not None: + normalized["user_property"] = list(user_properties) + return normalized + + +def topic_matches_sub(topic_filter: str, topic: str) -> bool: + filter_parts = topic_filter.split("/") + topic_parts = topic.split("/") + for index, part in enumerate(filter_parts): + if part == "#": + return index == len(filter_parts) - 1 + if index >= len(topic_parts): + return False + if part != "+" and part != topic_parts[index]: + return False + return len(topic_parts) == len(filter_parts) + + +def _split_host_port(broker: str) -> tuple[str, int]: + if ":" not in broker: + return broker, 1883 + host, port = broker.rsplit(":", 1) + try: + return host, int(port) + except ValueError: + return broker, 1883 + + +class Client: + def __init__( + self, + broker: str, + *, + protocol: int = MQTTv5, + client_id: str | None = None, + keepalive: int = 60, + ): + self.broker = broker + self.protocol = protocol + self.client_id = client_id or f"miniconf-py-{uuid.uuid4().hex}" + self.keepalive = keepalive + self.messages = _Messages() + self._client: GmqttClient | None = None + self._subacks: dict[int, asyncio.Future[tuple[int, ...]]] = {} + self._unsubacks: dict[int, asyncio.Future[tuple[int, ...]]] = {} + + async def __aenter__(self) -> Client: + host, port = _split_host_port(self.broker) + client = GmqttClient(self.client_id, clean_session=True) + client.on_message = self._on_message + client.on_subscribe = self._on_subscribe + client.on_unsubscribe = self._on_unsubscribe + await client.connect( + host, port, keepalive=self.keepalive, version=self.protocol + ) + self._client = client + return self + + async def __aexit__(self, *_exc_info) -> None: + await self.disconnect() + + async def disconnect(self) -> None: + if self._client is None: + return + client = self._client + self._client = None + await client.disconnect() + + async def subscribe( + self, + topic_filter: str, + *, + qos: int = 1, + no_local: bool = False, + retain_as_published: bool = False, + retain_handling: int = 0, + ) -> None: + client = self._require_client() + mid = client.subscribe( + GmqttSubscription( + topic_filter, + qos=qos, + no_local=no_local, + retain_as_published=retain_as_published, + retain_handling_options=retain_handling, + ) + ) + if mid is not None: + fut = asyncio.get_running_loop().create_future() + self._subacks[mid] = fut + reasons = await asyncio.wait_for(fut, 3.0) + if not reasons or reasons[0] >= 128: + raise MQTTError(f"SUBACK failed for {topic_filter}: {reasons}") + + async def unsubscribe(self, topic_filter: str) -> None: + client = self._require_client() + mid = client.unsubscribe(topic_filter) + if mid is not None: + fut = asyncio.get_running_loop().create_future() + self._unsubacks[mid] = fut + reasons = await asyncio.wait_for(fut, 3.0) + if reasons and reasons[0] >= 128: + raise MQTTError(f"UNSUBACK failed for {topic_filter}: {reasons}") + + async def publish( + self, + topic: str, + *, + payload: str | bytes, + qos: int = 1, + retain: bool = False, + properties: dict[str, Any] | None = None, + ) -> None: + self._require_client().publish( + topic, + payload, + qos=qos, + retain=retain, + **(properties or {}), + ) + await asyncio.sleep(0) + + def _require_client(self) -> GmqttClient: + if self._client is None: + raise MQTTError("MQTT client is not connected") + return self._client + + def _on_message( + self, + _client: GmqttClient, + topic: str, + payload: bytes, + _qos: int, + properties: dict[str, Any], + ) -> None: + self.messages.put( + Message( + topic, + payload, + retain=bool(properties.get("retain")), + properties=_normalize_properties(properties), + ) + ) + + def _on_subscribe( + self, + _client: GmqttClient, + mid: int, + reasons: tuple[int, ...], + _properties: dict[str, Any], + ) -> None: + if fut := self._subacks.pop(mid, None): + fut.set_result(reasons) + + def _on_unsubscribe( + self, + _client: GmqttClient, + mid: int, + reasons: tuple[int, ...], + ) -> None: + if fut := self._unsubacks.pop(mid, None): + fut.set_result(reasons) diff --git a/py/miniconf/_ops.py b/py/miniconf/_ops.py index 0dc28cdb..751a46d5 100644 --- a/py/miniconf/_ops.py +++ b/py/miniconf/_ops.py @@ -6,32 +6,23 @@ import json from typing import TYPE_CHECKING, Any -from aiomqtt import Client, Message - from .common import ( + RETAINED_SUBSCRIPTION, + AliveManifest, LOGGER, - MM2_PROTO, BurstState, MiniconfException, - is_authoritative, is_retained, + alive_manifest, quiet_window, - retained_options, settings_topics, - subtree_match, ) +from ._mqtt import Client if TYPE_CHECKING: from .client import Miniconf -def _properties(message: Message) -> dict[str, Any]: - try: - return message.properties.json() - except AttributeError: - return {} - - async def discover( client: Client, prefix: str, @@ -45,7 +36,14 @@ async def discover( topic = f"{prefix}{suffix}" start = asyncio.get_running_loop().time() - await client.subscribe(topic, options=retained_options()) + qos, no_local, retain_as_published, retain_handling = RETAINED_SUBSCRIPTION + await client.subscribe( + topic, + qos=qos, + no_local=no_local, + retain_as_published=retain_as_published, + retain_handling=retain_handling, + ) quiet = quiet_window( start, asyncio.get_running_loop().time(), @@ -67,15 +65,15 @@ async def listen(): return if not is_retained(message): continue - peer = message.topic.value.removesuffix(suffix) + peer = message.topic.removesuffix(suffix) + if not message.payload: + discovered.pop(peer, None) + continue try: - manifest = json.loads(message.payload) - except json.JSONDecodeError: + manifest = alive_manifest(json.loads(message.payload)) + except (json.JSONDecodeError, MiniconfException): LOGGER.info("Ignoring %s not/invalid alive", peer) continue - if not isinstance(manifest, dict) or manifest.get("proto") != MM2_PROTO: - LOGGER.info("Ignoring %s unsupported protocol alive", peer) - continue discovered[peer] = manifest deadline = asyncio.get_running_loop().time() + quiet @@ -86,11 +84,11 @@ async def listen(): return discovered -async def _manifest(interface: Miniconf, *, timeout: float = 3.0) -> dict[str, Any]: +async def _manifest(interface: Miniconf, *, timeout: float = 3.0) -> AliveManifest: if interface._manifest is not None: return interface._manifest async with interface._watch( - f"{interface.prefix}/alive", retained_options() + f"{interface.prefix}/alive", RETAINED_SUBSCRIPTION ) as queue: end = asyncio.get_running_loop().time() + timeout while True: @@ -119,7 +117,7 @@ async def _collect_retained_settings( start = asyncio.get_running_loop().time() retained: dict[str, Any] = {} (topic_filter,) = settings_topics(interface.prefix, root) - async with interface._watch(topic_filter, retained_options()) as queue: + async with interface._watch(topic_filter, RETAINED_SUBSCRIPTION) as queue: now = asyncio.get_running_loop().time() burst = BurstState.from_roundtrip(start, now, rel_timeout, abs_timeout) end = now + timeout @@ -138,18 +136,13 @@ async def _collect_retained_settings( continue if not is_retained(message): continue - topic = message.topic.value - if not topic.startswith(f"{interface.prefix}/settings"): - continue - if not is_authoritative(_properties(message)): + event = interface._setting_event(message, root) + if event is None: continue - settings_path = topic.removeprefix(f"{interface.prefix}/settings") - if not subtree_match(settings_path, root): - continue - if not message.payload: - retained.pop(settings_path, None) + if not event.present: + retained.pop(event.path, None) else: - retained[settings_path] = json.loads(message.payload) + retained[event.path] = event.value burst.reset(asyncio.get_running_loop().time()) @@ -163,7 +156,7 @@ async def _collect_retained_topics( ) -> list[str]: start = asyncio.get_running_loop().time() seen: set[str] = set() - async with interface._watch(topic_filter, retained_options()) as queue: + async with interface._watch(topic_filter, RETAINED_SUBSCRIPTION) as queue: now = asyncio.get_running_loop().time() burst = BurstState.from_roundtrip(start, now, rel_timeout, abs_timeout) end = now + timeout @@ -183,7 +176,7 @@ async def _collect_retained_topics( if not is_retained(message): continue if message.payload: - seen.add(message.topic.value) + seen.add(message.topic) burst.reset(asyncio.get_running_loop().time()) @@ -197,11 +190,11 @@ async def _prune_schema( """Clear retained schema pages above the current manifest page count.""" manifest = await _manifest(interface, timeout=timeout) - pages = int(manifest["pages"]) + pages = manifest.pages seen: set[int] = set() start = asyncio.get_running_loop().time() async with interface._watch( - f"{interface.prefix}/schema/#", retained_options() + f"{interface.prefix}/schema/#", RETAINED_SUBSCRIPTION ) as queue: now = asyncio.get_running_loop().time() burst = BurstState.from_roundtrip(start, now, rel_timeout, abs_timeout) @@ -220,7 +213,7 @@ async def _prune_schema( break if not is_retained(message): continue - suffix = message.topic.value.removeprefix(f"{interface.prefix}/schema/") + suffix = message.topic.removeprefix(f"{interface.prefix}/schema/") try: seen.add(int(suffix)) except ValueError: @@ -260,7 +253,6 @@ async def _prune_settings( qos=1, retain=True, ) - interface._settings.pop(cache_path, None) return stale @@ -285,5 +277,4 @@ async def force_prune(interface: Miniconf, *, timeout: float = 3.0) -> list[str] await interface.client.publish(topic, payload=b"", qos=1, retain=True) interface._schema = None interface._manifest = None - interface._settings.clear() return [topic.removeprefix(f"{interface.prefix}/") for topic in topics] diff --git a/py/miniconf/cli.py b/py/miniconf/cli.py index f6f3ecf0..83104138 100644 --- a/py/miniconf/cli.py +++ b/py/miniconf/cli.py @@ -9,10 +9,9 @@ import os import sys -from aiomqtt import Client - from .client import Miniconf, RawMiniconf -from .common import LOGGER, MQTTv5, MiniconfException, json_dumps, validate_path +from .common import LOGGER, MiniconfException, json_dumps, validate_path +from ._mqtt import Client from ._ops import discover, force_prune, prune from .render import render_schema_tree, render_value_tree @@ -60,7 +59,7 @@ async def _main() -> None: level=logging.WARN - 10 * args.verbose, ) - async with Client(args.broker, protocol=MQTTv5) as client: + async with Client(args.broker) as client: prefix = await _resolve_prefix(client, args.prefix, args.discover) if args.raw and (args.prune or args.force_prune): raise MiniconfException( @@ -207,14 +206,16 @@ def normalize(path: str) -> str: print(render_schema_tree(await interface.schema(timeout=timeout), path)) elif arg.endswith("!!"): path = normalize(arg.removesuffix("!!")) - async with interface.track(path, timeout=timeout) as tracked: - for dump_path, value in sorted(tracked.snapshot().items()): - print(f"{dump_path}={json_dumps(value)}") + for dump_path, value in sorted( + (await interface.snapshot(path, timeout=timeout)).items() + ): + print(f"{dump_path}={json_dumps(value)}") elif arg.endswith("!"): path = normalize(arg.removesuffix("!")) schema = await interface.schema(timeout=timeout) - async with interface.track(path, timeout=timeout) as tracked: - print(render_value_tree(schema, tracked.snapshot(), tracked.root)) + root = schema.path(path) + snapshot = await interface.snapshot(root, timeout=timeout) + print(render_value_tree(schema, snapshot, root)) elif "=" in arg: path, value = arg.split("=", 1) path, base = _normalize_command_path(path, base, subtree=False) diff --git a/py/miniconf/client.py b/py/miniconf/client.py index 84e157f3..da675642 100644 --- a/py/miniconf/client.py +++ b/py/miniconf/client.py @@ -11,58 +11,77 @@ from dataclasses import dataclass from typing import Any -import paho.mqtt.client as mqtt -from aiomqtt import Client, Message, MqttError -from paho.mqtt.properties import PacketTypes, Properties -from paho.mqtt.subscribeoptions import SubscribeOptions - from . import _ops from .common import ( + DEFAULT_SUBSCRIPTION, LOGGER, - MM2_PROTO, + RETAINED_SUBSCRIPTION, + AliveManifest, + SubscriptionKey, BurstState, MiniconfException, + alive_manifest, is_authoritative, is_retained, json_dumps, message_expiry, - retained_options, settings_topics, - subscription_key, subtree_match, validate_path, ) +from ._mqtt import Client, Message, MQTTError, topic_matches_sub from .schema import Schema def _properties(message: Message) -> dict[str, Any]: - try: - return message.properties.json() - except AttributeError: - return {} + return message.properties def _response_code(properties: dict[str, Any]) -> str: try: - return dict(properties["UserProperty"])["code"] + return dict(properties["user_property"])["code"] except KeyError as exc: raise MiniconfException("Protocol", "Missing response code") from exc def _response_cd(properties: dict[str, Any]) -> bytes | None: - try: - return bytes.fromhex(properties["CorrelationData"]) - except KeyError: + return properties.get("correlation_data") + + +@dataclass(frozen=True) +class SettingEvent: + """One authoritative `/settings` publication.""" + + path: str + present: bool + retained: bool = True + value: Any = None + rev: str | None = None + + +def _setting_event(message: Message, prefix: str, root: str) -> SettingEvent | None: + properties = _properties(message) + if not is_retained(message) or not is_authoritative(properties): + return None + topic = message.topic + if not topic.startswith(f"{prefix}/settings"): return None + path = topic.removeprefix(f"{prefix}/settings") + if path and not path.startswith("/"): + return None + if not subtree_match(path, root): + return None + rev = _user_property(properties, "rev") + if not message.payload: + return SettingEvent(path, False, rev=rev) + return SettingEvent(path, True, value=json.loads(message.payload), rev=rev) -@dataclass -class _TrackState: - burst: BurstState - timeout: float - rel_timeout: float - abs_timeout: float - reload_task: asyncio.Task[None] | None = None +def _user_property(properties: dict[str, Any], name: str) -> str | None: + try: + return dict(properties.get("user_property", ())).get(name) + except (TypeError, ValueError): + return None class _BaseClient: @@ -72,15 +91,15 @@ def __init__(self, client: Client, prefix: str): self.response_topic = f"{prefix}/response/{uuid.uuid4().hex}" self._inflight: dict[bytes, asyncio.Future[None]] = {} self._watchers: dict[str, list[asyncio.Queue[Message]]] = defaultdict(list) - self._subscriptions: dict[str, tuple[int, tuple[int, bool, bool, int]]] = {} + self._subscriptions: dict[str, tuple[int, SubscriptionKey]] = {} self._listener = asyncio.create_task(self._listen()) self._subscribed = asyncio.Event() def _listen_topics(self) -> tuple[str, ...]: return (self.response_topic,) - def _listen_options(self, _topic: str) -> SubscribeOptions | None: - return None + def _listen_subscription(self, _topic: str) -> SubscriptionKey: + return DEFAULT_SUBSCRIPTION async def __aenter__(self): return self @@ -104,30 +123,30 @@ async def _listen(self): topics: list[str] = [] try: for topic in self._listen_topics(): - await self._subscribe(topic, self._listen_options(topic)) + await self._subscribe(topic, self._listen_subscription(topic)) topics.append(topic) self._subscribed.set() async for message in self.client.messages: self._dispatch(message) except asyncio.CancelledError: pass - except MqttError: + except MQTTError: LOGGER.debug("MQTT error", exc_info=True) finally: self._subscribed.clear() for topic in reversed(topics): try: await self._unsubscribe(topic, missing_ok=True) - except MqttError: + except MQTTError: LOGGER.debug("MQTT unsubscribe error", exc_info=True) def _dispatch(self, message: Message): - topic = message.topic.value + topic = message.topic properties = _properties(message) LOGGER.debug("Received %s: %s [%s]", topic, message.payload, properties) for topic_filter, queues in tuple(self._watchers.items()): - if mqtt.topic_matches_sub(topic_filter, topic): + if topic_matches_sub(topic_filter, topic): for queue in tuple(queues): queue.put_nowait(message) @@ -139,11 +158,11 @@ def _dispatch(self, message: Message): def _handle_response(self, properties: dict[str, Any], payload: bytes): cd = _response_cd(properties) if cd is None: - LOGGER.debug("Discarding response without CorrelationData") + LOGGER.debug("Discarding response without correlation_data") return fut = self._inflight.pop(cd, None) if fut is None: - LOGGER.debug("Discarding unexpected CorrelationData: %s", cd.hex()) + LOGGER.debug("Discarding unexpected correlation_data: %s", cd.hex()) return if fut.done(): LOGGER.debug("Discarding late response: %s", cd.hex()) @@ -160,22 +179,27 @@ def _handle_message( pass async def _subscribe( - self, topic_filter: str, options: SubscribeOptions | None = None + self, + topic_filter: str, + subscription: SubscriptionKey = DEFAULT_SUBSCRIPTION, ): - key = subscription_key(options) existing = self._subscriptions.get(topic_filter) if existing is None: - if options is None: - await self.client.subscribe(topic_filter, qos=1) - else: - await self.client.subscribe(topic_filter, options=options) + qos, no_local, retain_as_published, retain_handling = subscription + await self.client.subscribe( + topic_filter, + qos=qos, + no_local=no_local, + retain_as_published=retain_as_published, + retain_handling=retain_handling, + ) LOGGER.debug("Subscribed to %s", topic_filter) - self._subscriptions[topic_filter] = (1, key) + self._subscriptions[topic_filter] = (1, subscription) return count, existing_key = existing - if existing_key != key: + if existing_key != subscription: raise MiniconfException("Subscription", topic_filter) - self._subscriptions[topic_filter] = (count + 1, key) + self._subscriptions[topic_filter] = (count + 1, subscription) async def _unsubscribe(self, topic_filter: str, *, missing_ok: bool = False): existing = self._subscriptions.get(topic_filter) @@ -194,13 +218,15 @@ async def _unsubscribe(self, topic_filter: str, *, missing_ok: bool = False): @asynccontextmanager async def _watch( - self, topic_filter: str, options: SubscribeOptions | None = None + self, + topic_filter: str, + subscription: SubscriptionKey = DEFAULT_SUBSCRIPTION, ) -> AsyncIterator[asyncio.Queue[Message]]: queue: asyncio.Queue[Message] = asyncio.Queue() watchers = self._watchers[topic_filter] watchers.append(queue) try: - await self._subscribe(topic_filter, options) + await self._subscribe(topic_filter, subscription) except Exception: watchers.remove(queue) if not watchers: @@ -214,18 +240,36 @@ async def _watch( del self._watchers[topic_filter] await self._unsubscribe(topic_filter) + async def _watch_settings(self, root: str) -> AsyncIterator[SettingEvent]: + async with self._settings_queue(root) as queue: + while True: + message = await queue.get() + event = self._setting_event(message, root) + if event is not None: + yield event + + def _setting_event(self, message: Message, root: str) -> SettingEvent | None: + return _setting_event(message, self.prefix, root) + + @asynccontextmanager + async def _settings_queue(self, root: str) -> AsyncIterator[asyncio.Queue[Message]]: + (topic_filter,) = settings_topics(self.prefix, root) + async with self._watch(topic_filter, RETAINED_SUBSCRIPTION) as queue: + yield queue + async def _publish_set( self, path: str, payload: str, *, response: bool, timeout: float | None = None ): - props = Properties(PacketTypes.PUBLISH) - props.PayloadFormatIndicator = 1 - props.MessageExpiryInterval = message_expiry(timeout) + props: dict[str, Any] = { + "payload_format_id": 1, + "message_expiry_interval": message_expiry(timeout), + } fut = None if response: await self._subscribed.wait() - props.ResponseTopic = self.response_topic + props["response_topic"] = self.response_topic cd = uuid.uuid4().bytes - props.CorrelationData = cd + props["correlation_data"] = cd fut = asyncio.get_running_loop().create_future() assert cd not in self._inflight self._inflight[cd] = fut @@ -248,7 +292,7 @@ async def _read_retained_json( *, timeout: float, ): - async with watch(topic_filter, retained_options()) as queue: + async with watch(topic_filter, RETAINED_SUBSCRIPTION) as queue: end = asyncio.get_running_loop().time() + timeout while True: remaining = end - asyncio.get_running_loop().time() @@ -271,183 +315,58 @@ async def _read_retained_json( ) from exc -class TrackedSubtree: - """Scoped retained-settings subtree tracker.""" - - def __init__( - self, - client: Miniconf, - path: str, - *, - timeout: float, - rel_timeout: float, - abs_timeout: float, - ): - self._client = client - self._path = path - self._timeout = timeout - self._rel_timeout = rel_timeout - self._abs_timeout = abs_timeout - self._root: str | None = None - - @property - def root(self) -> str: - if self._root is None: - raise MiniconfException("Closed", "Tracked subtree is closed") - return self._root - - async def __aenter__(self) -> TrackedSubtree: - self._root = await self._client._open_track( - self._path, - timeout=self._timeout, - rel_timeout=self._rel_timeout, - abs_timeout=self._abs_timeout, - ) - return self - - async def __aexit__(self, *_exc_info) -> None: - await self.close() - - async def close(self) -> None: - if self._root is None: - return - await self._client._close_track(self._root) - self._root = None - - @property - def ready(self) -> bool: - """Whether the tracked cache is currently usable.""" - - if self._root is None: - return False - task = self._client._tracking_reload() - if task is None: - return True - if not task.done() or task.cancelled(): - return False - return task.exception() is None - - async def wait_ready(self, timeout: float | None = None) -> None: - """Wait until any tracked-cache reload triggered by `/alive` is complete.""" - - if self._root is None: - raise MiniconfException("Closed", "Tracked subtree is closed") - await self._client._wait_tracking_reload(timeout) - - def _resolve(self, path: str, *, leaf: bool) -> str: - root = self.root - if not path: - full = root - else: - full = validate_path(path) - full = self._client._schema.path(full) - if not subtree_match(full, root): - raise MiniconfException("Untracked", full) - if leaf and self._client._schema.node(full).kind != "leaf": - raise MiniconfException("LeafRequired", full) - return full - - def cached(self, path: str = ""): - """Read one cached tracked leaf.""" - - path = self._resolve(path, leaf=True) - self._client._check_tracking_reload() - try: - return self._client._settings[path] - except KeyError as exc: - raise MiniconfException("NotFound", path) from exc - - def snapshot(self, path: str = "") -> dict[str, Any]: - """Return cached retained values below one tracked subtree.""" - - root = self._resolve(path, leaf=False) - self._client._check_tracking_reload() - return { - cache_path: value - for cache_path, value in self._client._settings.items() - if subtree_match(cache_path, root) - } - - class Miniconf(_BaseClient): - """Long-lived Miniconf session with schema and settings caches. + """Long-lived Miniconf session with retained schema cache. The client keeps `/alive` subscribed to notice new device epochs and schema revisions. Retained - `settings/#` publications without `auth` are treated as non-authoritative and ignored. One - explicit tracked settings subtree may be active at a time through `track()`. + `settings/#` publications without `auth` are treated as non-authoritative and ignored. """ def __init__(self, client: Client, prefix: str): self.alive_topic = f"{prefix}/alive" self._schema: Schema | None = None - self._manifest: dict[str, Any] | None = None - self._settings: dict[str, Any] = {} - self._tracked_root: str | None = None - self._track_state: _TrackState | None = None + self._manifest: AliveManifest | None = None super().__init__(client, prefix) + @classmethod + @asynccontextmanager + async def connect( + cls, broker: str, prefix: str, **client_kwargs: Any + ) -> AsyncIterator[Miniconf]: + async with Client(broker, **client_kwargs) as client: + async with cls(client, prefix) as interface: + yield interface + def _listen_topics(self) -> tuple[str, ...]: return self.response_topic, self.alive_topic - def _listen_options(self, topic: str) -> SubscribeOptions | None: + def _listen_subscription(self, topic: str) -> SubscriptionKey: if topic == self.alive_topic: - return retained_options() - return None - - async def close(self) -> None: - if self._tracked_root is not None: - await self._close_track(self._tracked_root) - await super().close() + return RETAINED_SUBSCRIPTION + return DEFAULT_SUBSCRIPTION def _handle_message(self, message: Message, topic: str, properties: dict[str, Any]): if topic == self.alive_topic: self._note_manifest_payload(message.payload) - return - if self._tracked_root is None or not topic.startswith( - f"{self.prefix}/settings" - ): - return - if not is_authoritative(properties) or not is_retained(message): - return - path = topic.removeprefix(f"{self.prefix}/settings") - if not subtree_match(path, self._tracked_root): - return - if not message.payload: - self._settings.pop(path, None) - else: - self._settings[path] = json.loads(message.payload) - now = asyncio.get_running_loop().time() - assert self._track_state is not None - self._track_state.burst.reset(now) - - def _note_manifest(self, manifest: dict[str, Any]): - if not isinstance(manifest, dict): + def _note_manifest(self, manifest: Any): + prev = self._manifest + try: + next_manifest = alive_manifest(manifest) + except MiniconfException: LOGGER.debug("Ignoring invalid alive manifest: %r", manifest) return - if manifest.get("proto") != MM2_PROTO: - LOGGER.debug("Ignoring unsupported alive manifest: %r", manifest) - return - - prev = self._manifest - self._manifest = manifest + self._manifest = next_manifest if prev is None: - prev_epoch = prev_schema_rev = None + prev_schema_rev = None else: - prev_epoch = prev.get("epoch") - prev_schema_rev = prev.get("schema_rev") - epoch = manifest.get("epoch") - schema_rev = manifest.get("schema_rev") - if prev_epoch != epoch or prev_schema_rev != schema_rev: - self._start_tracking_reload() - if prev_schema_rev != schema_rev: + prev_schema_rev = prev.schema_rev + if prev_schema_rev != next_manifest.schema_rev: self._schema = None def _note_device_gone(self): self._manifest = None self._schema = None - self._settings.clear() - self._cancel_tracking_reload() def _note_manifest_payload(self, payload: bytes): if not payload: @@ -491,17 +410,46 @@ async def get(self, path: str, *, timeout: float = 3.0): timeout=timeout, ) + async def snapshot( + self, + path: str = "", + *, + timeout: float = 3.0, + rel_timeout: float = 3.0, + abs_timeout: float = 0.1, + ) -> dict[str, Any]: + """Return a finite retained settings snapshot below one subtree.""" + + return await _ops._collect_retained_settings( + self, + path, + timeout=timeout, + rel_timeout=rel_timeout, + abs_timeout=abs_timeout, + ) + + async def watch( + self, path: str = "", *, timeout: float = 3.0 + ) -> AsyncIterator[SettingEvent]: + """Yield authoritative settings updates below one subtree without waiting for quiescence.""" + + root = (await self.schema(timeout=timeout)).path(path) + async for event in self._watch_settings(root): + yield event + async def schema(self, *, timeout: float = 3.0) -> Schema: """Load and cache the retained paged schema.""" if self._schema is not None: return self._schema manifest = await _ops._manifest(self, timeout=timeout) - schema_rev = int(manifest["schema_rev"]) - pages = int(manifest["pages"]) + schema_rev = manifest.schema_rev + pages = manifest.pages defs: list[list[dict[str, Any]] | None] = [None] * pages - async with self._watch(f"{self.prefix}/schema/#", retained_options()) as queue: + async with self._watch( + f"{self.prefix}/schema/#", RETAINED_SUBSCRIPTION + ) as queue: deadline = asyncio.get_running_loop().time() + timeout while any(page is None for page in defs): remaining = deadline - asyncio.get_running_loop().time() @@ -510,7 +458,7 @@ async def schema(self, *, timeout: float = 3.0) -> Schema: message = await asyncio.wait_for(queue.get(), remaining) if not is_retained(message): continue - suffix = message.topic.value.removeprefix(f"{self.prefix}/schema/") + suffix = message.topic.removeprefix(f"{self.prefix}/schema/") try: page = int(suffix) except ValueError: @@ -525,156 +473,6 @@ async def schema(self, *, timeout: float = 3.0) -> Schema: ) return self._schema - def track( - self, - path: str = "", - *, - timeout: float = 3.0, - rel_timeout: float = 3.0, - abs_timeout: float = 0.1, - ) -> TrackedSubtree: - """Return a scoped tracker for one retained settings subtree.""" - - return TrackedSubtree( - self, - path, - timeout=timeout, - rel_timeout=rel_timeout, - abs_timeout=abs_timeout, - ) - - async def _open_track( - self, - path: str, - *, - timeout: float, - rel_timeout: float, - abs_timeout: float, - ) -> str: - # Match the embedded retained-load phase: measure the subscribe round trip once, then wait - # for quiescence until the last retained publication plus abs_timeout + rel_timeout * rtt. - if self._tracked_root is not None: - raise MiniconfException("Tracked", self._tracked_root) - root = (await self.schema(timeout=timeout)).path(path) - start = asyncio.get_running_loop().time() - self._tracked_root = root - self._track_state = _TrackState( - burst=BurstState.from_roundtrip(start, start, rel_timeout, abs_timeout), - timeout=timeout, - rel_timeout=rel_timeout, - abs_timeout=abs_timeout, - ) - self._settings.clear() - try: - for topic_filter in settings_topics(self.prefix, root): - await self._subscribe(topic_filter, retained_options()) - now = asyncio.get_running_loop().time() - assert self._track_state is not None - self._track_state.burst.set_roundtrip(start, now, rel_timeout, abs_timeout) - await self._wait_tracking(timeout) - except Exception: - for topic_filter in settings_topics(self.prefix, root): - await self._unsubscribe(topic_filter) - self._tracked_root = None - self._track_state = None - self._settings.clear() - raise - return root - - async def _close_track(self, root: str): - if self._tracked_root != root: - raise MiniconfException("Tracked", root) - self._cancel_tracking_reload() - for topic_filter in settings_topics(self.prefix, root): - await self._unsubscribe(topic_filter) - self._tracked_root = None - self._track_state = None - self._settings.clear() - - def _start_tracking_reload(self) -> None: - state = self._track_state - if self._tracked_root is None or state is None: - self._settings.clear() - return - if state.reload_task is not None and not state.reload_task.done(): - return - self._settings.clear() - state.reload_task = asyncio.create_task(self._reload_tracking()) - - def _cancel_tracking_reload(self) -> None: - state = self._track_state - if state is None or state.reload_task is None: - return - state.reload_task.cancel() - state.reload_task = None - - def _check_tracking_reload(self) -> None: - task = self._tracking_reload() - if task is None: - return - if task.done(): - self._finish_tracking_reload(task) - return - raise MiniconfException("Reloading", self._tracked_root or "/") - - def _tracking_reload(self) -> asyncio.Task[None] | None: - state = self._track_state - if state is None or state.reload_task is None: - return None - return state.reload_task - - def _finish_tracking_reload(self, task: asyncio.Task[None]) -> None: - try: - task.result() - finally: - state = self._track_state - if state is not None and state.reload_task is task: - state.reload_task = None - - async def _wait_tracking_reload(self, timeout: float | None) -> None: - task = self._tracking_reload() - if task is None: - return - await asyncio.wait_for(asyncio.shield(task), timeout) - self._finish_tracking_reload(task) - - async def _reload_tracking(self) -> None: - state = self._track_state - root = self._tracked_root - if state is None or root is None: - return - if self._schema is None: - root = (await self.schema(timeout=state.timeout)).path(root) - self._tracked_root = root - start = asyncio.get_running_loop().time() - for topic_filter in settings_topics(self.prefix, root): - # Force a retained replay for the already-open tracked subscription without changing - # the local refcount: a new alive generation means the cache must be rebuilt from the - # authoritative retained mirror. - await self.client.subscribe(topic_filter, options=retained_options()) - state.burst.reset(asyncio.get_running_loop().time()) - state.burst.set_roundtrip( - start, - asyncio.get_running_loop().time(), - rel_timeout=state.rel_timeout, - abs_timeout=state.abs_timeout, - ) - await self._wait_tracking(state.timeout) - - async def _wait_tracking(self, timeout: float): - state = self._track_state - assert state is not None - end = asyncio.get_running_loop().time() + timeout - while True: - now = asyncio.get_running_loop().time() - if now >= state.burst.deadline: - return - if now >= end: - raise TimeoutError( - f"Timed out waiting for retained settings under {self._tracked_root or '/'}" - ) - await asyncio.sleep(min(0.01, state.burst.deadline - now, end - now)) - class RawMiniconf(_BaseClient): """Schema-less Miniconf client for exact-path GET and SET operations.""" @@ -682,6 +480,15 @@ class RawMiniconf(_BaseClient): def __init__(self, client: Client, prefix: str): super().__init__(client, prefix) + @classmethod + @asynccontextmanager + async def connect( + cls, broker: str, prefix: str, **client_kwargs: Any + ) -> AsyncIterator[RawMiniconf]: + async with Client(broker, **client_kwargs) as client: + async with cls(client, prefix) as interface: + yield interface + async def set( self, path: str, @@ -707,3 +514,50 @@ async def get(self, path: str, *, timeout: float = 3.0): path, timeout=timeout, ) + + async def snapshot( + self, + path: str = "", + *, + timeout: float = 3.0, + rel_timeout: float = 3.0, + abs_timeout: float = 0.1, + ) -> dict[str, Any]: + """Return a finite retained settings snapshot below one exact subtree.""" + + root = validate_path(path) + start = asyncio.get_running_loop().time() + retained: dict[str, Any] = {} + async with self._settings_queue(root) as queue: + burst = BurstState.from_roundtrip( + start, + asyncio.get_running_loop().time(), + rel_timeout, + abs_timeout, + ) + end = asyncio.get_running_loop().time() + timeout + while True: + now = asyncio.get_running_loop().time() + if now >= burst.deadline or now >= end: + return retained + try: + message = await asyncio.wait_for( + queue.get(), min(burst.deadline, end) - now + ) + except TimeoutError: + continue + event = self._setting_event(message, root) + if event is None: + continue + if not event.present: + retained.pop(event.path, None) + else: + retained[event.path] = event.value + burst.reset(asyncio.get_running_loop().time()) + + async def watch(self, path: str = "") -> AsyncIterator[SettingEvent]: + """Yield authoritative settings updates below one exact subtree.""" + + root = validate_path(path) + async for event in self._watch_settings(root): + yield event diff --git a/py/miniconf/common.py b/py/miniconf/common.py index 8ce24404..d33a65d4 100644 --- a/py/miniconf/common.py +++ b/py/miniconf/common.py @@ -1,18 +1,19 @@ """Common code for the Miniconf MQTT clients.""" from dataclasses import dataclass +from typing import Any import json import logging -import paho.mqtt -from paho.mqtt.subscribeoptions import SubscribeOptions - -MQTTv5 = paho.mqtt.enums.MQTTProtocolVersion.MQTTv5 MM2_PROTO = 1 LOGGER = logging.getLogger("miniconf") # Expire transient set requests. Retained alive/schema/settings publications are storage. TRANSIENT_EXPIRY_S = 30 +RETAIN_SEND_ON_SUBSCRIBE = 0 +SubscriptionKey = tuple[int, bool, bool, int] +DEFAULT_SUBSCRIPTION: SubscriptionKey = (1, False, False, RETAIN_SEND_ON_SUBSCRIBE) +RETAINED_SUBSCRIPTION: SubscriptionKey = (1, False, True, RETAIN_SEND_ON_SUBSCRIBE) def message_expiry(timeout: float | None) -> int: @@ -21,23 +22,29 @@ def message_expiry(timeout: float | None) -> int: return max(1, int(timeout + 0.999)) -def retained_options() -> SubscribeOptions: - return SubscribeOptions( - qos=1, - retainAsPublished=True, - retainHandling=SubscribeOptions.RETAIN_SEND_ON_SUBSCRIBE, - ) - - -def subscription_key(options: SubscribeOptions | None) -> tuple[int, bool, bool, int]: - if options is None: - options = SubscribeOptions(qos=1) - return ( - options.QoS, - options.noLocal, - options.retainAsPublished, - options.retainHandling, - ) +@dataclass(frozen=True) +class AliveManifest: + proto: int + epoch: int + schema_rev: int + pages: int + + +def alive_manifest(value: Any) -> AliveManifest: + if not isinstance(value, dict): + raise MiniconfException("Protocol", "Invalid alive manifest") + if value.get("proto") != MM2_PROTO: + raise MiniconfException("Protocol", "Unsupported alive manifest") + epoch = value.get("epoch") + schema_rev = value.get("schema_rev") + pages = value.get("pages") + if ( + not isinstance(epoch, int) + or not isinstance(schema_rev, int) + or not isinstance(pages, int) + ): + raise MiniconfException("Protocol", "Invalid alive manifest") + return AliveManifest(MM2_PROTO, epoch, schema_rev, pages) def is_retained(message) -> bool: @@ -45,7 +52,7 @@ def is_retained(message) -> bool: def user_property_values(properties: dict, name: str) -> list[str]: - return [value for key, value in properties.get("UserProperty", ()) if key == name] + return [value for key, value in properties.get("user_property", ()) if key == name] def is_authoritative(properties: dict) -> bool: diff --git a/py/pyproject.toml b/py/pyproject.toml index 59335934..002255ac 100644 --- a/py/pyproject.toml +++ b/py/pyproject.toml @@ -14,7 +14,7 @@ authors = [ { name = "Ryan Summers", email = "ryan.summers@vertigo-designs.com" }, { name = "Robert Jördens", email = "rj@quartiq.de" }, ] -dependencies = ["aiomqtt>=2.1"] +dependencies = ["gmqtt>=0.7"] classifiers = [ "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.11", diff --git a/py/test.py b/py/test.py index b0b10e5b..6adc63e5 100644 --- a/py/test.py +++ b/py/test.py @@ -13,10 +13,10 @@ from queue import Empty, Queue import paho.mqtt.client as mqtt -from aiomqtt import Client from miniconf.client import Miniconf, RawMiniconf from miniconf.cli import _normalize_command_path -from miniconf.common import MQTTv5, MiniconfException +from miniconf.common import MiniconfException +from miniconf._mqtt import Client from miniconf.render import render_schema_tree, render_value_tree from miniconf.schema import Indices, Packed, Schema, SchemaNode @@ -191,30 +191,46 @@ async def test_listener_close_tolerates_released_subscription() -> None: async def close_client(client: Client, timeout: float = 1.0) -> None: - """Bound aiomqtt teardown so the harness cannot hang on disconnect.""" + """Bound MQTT teardown so the harness cannot hang on disconnect.""" - try: - await asyncio.wait_for(client.__aexit__(None, None, None), timeout) - except TimeoutError: - client._client.disconnect() # type: ignore[attr-defined] + await asyncio.wait_for(client.__aexit__(None, None, None), timeout) -async def wait_cached(tracked, path: str, expected, timeout: float = 3.0): +async def wait_event_value(events, path: str, expected, timeout: float = 3.0): + end = asyncio.get_running_loop().time() + timeout + while True: + remaining = end - asyncio.get_running_loop().time() + if remaining <= 0: + raise TimeoutError(f"Timed out waiting for event {path}={expected!r}") + event = await asyncio.wait_for(events.__anext__(), remaining) + if event.path == path and event.present and event.value == expected: + return event + + +async def wait_event_delete(events, path: str, timeout: float = 3.0): end = asyncio.get_running_loop().time() + timeout while True: - with contextlib.suppress(TimeoutError): - await tracked.wait_ready(max(0.0, end - asyncio.get_running_loop().time())) - try: - value = tracked.cached(path) - except MiniconfException: - value = object() - if value == expected: - return remaining = end - asyncio.get_running_loop().time() if remaining <= 0: - raise TimeoutError( - f"Timed out waiting for cached {path or '/'}={expected!r}" - ) + raise TimeoutError(f"Timed out waiting for delete event {path}") + event = await asyncio.wait_for(events.__anext__(), remaining) + if event.path == path and not event.present: + return event + + +async def wait_snapshot_value( + client, root: str, path: str, expected, timeout: float = 3.0 +): + end = asyncio.get_running_loop().time() + timeout + while True: + snapshot = await client.snapshot( + root, timeout=max(0.1, end - asyncio.get_running_loop().time()) + ) + if snapshot.get(path, object()) == expected: + return snapshot + remaining = end - asyncio.get_running_loop().time() + if remaining <= 0: + raise TimeoutError(f"Timed out waiting for snapshot {path}={expected!r}") await asyncio.sleep(min(0.01, remaining)) @@ -314,7 +330,7 @@ async def main() -> None: settings.drain() schema_topics.drain() - client = Client(BROKER, protocol=MQTTv5) + client = Client(BROKER) await client.__aenter__() try: mc = Miniconf(client, TARGET) @@ -365,52 +381,86 @@ async def main() -> None: assert err.code == "NotFound", err else: raise AssertionError("expected lookup error for '/'") - async with mc.track(ENABLED) as tracked: - assert tracked.cached() is True + connected_value = None + async with Miniconf.connect(BROKER, TARGET) as connected: + connected_value = await connected.get(ENABLED) + assert connected_value is True + snapshot = await mc.snapshot(CONTROL) + assert snapshot[ENABLED] is True + assert snapshot[MODE] == "Run" try: - async with mc.track(CONTROL) as tracked: - tracked.cached() + await mc.get(CONTROL) except MiniconfException as err: assert err.code == "LeafRequired", err else: raise AssertionError(f"expected leaf-required error for {CONTROL}") - async with mc.track(CONTROL) as tracked: - assert tracked.cached(ENABLED) is True - dump = tracked.snapshot() - assert dump[ENABLED] is True - assert dump[MODE] == "Run" - assert render_value_tree(schema, dump, tracked.root).splitlines() == [ - "control", - "├─ enabled = true", - '└─ mode = "Run"', - ] - try: - async with mc.track(ENABLED): - raise AssertionError("expected tracked-subtree conflict") - except MiniconfException as err: - assert err.code == "Tracked", err + dump = await mc.snapshot(CONTROL) + assert dump[ENABLED] is True + assert dump[MODE] == "Run" + assert render_value_tree(schema, dump, CONTROL).splitlines() == [ + "control", + "├─ enabled = true", + '└─ mode = "Run"', + ] await mc.set(ENABLED, True) - async with mc.track(ENABLED) as tracked: - assert tracked.cached() is True + assert await mc.get(ENABLED) is True settings.drain() await mc.set(ENABLED, False, response=False) assert settings.wait_payload(3.0, f"{TARGET}/settings{ENABLED}") == "false" - async with mc.track("/output") as tracked: - assert tracked.cached(DAC0) == 1024 + events = mc.watch(ENABLED) + try: + await wait_event_value(events, ENABLED, False) + await mc.set(ENABLED, True) + await wait_event_value(events, ENABLED, True) + await mc.set(ENABLED, False) + await wait_event_value(events, ENABLED, False) + finally: + await events.aclose() + output = await mc.snapshot("/output") + assert output[DAC0] == 1024 + events = mc.watch("/output") + try: + await wait_event_value(events, DAC0, 1024) await mc.set(DAC0, 2048) - await wait_cached(tracked, DAC0, 2048) - tree_dump = tracked.snapshot() - assert tree_dump[DAC0] == 2048, tree_dump + await wait_event_value(events, DAC0, 2048) + finally: + await events.aclose() + await wait_snapshot_value(mc, "/output", DAC0, 2048) finally: await close_client(client) client = None - client = await Client(BROKER, protocol=MQTTv5).__aenter__() + client = await Client(BROKER).__aenter__() raw = RawMiniconf(client, TARGET) try: assert await raw.get(ENABLED) is False assert await raw.get(DAC0) == 2048 + assert (await raw.snapshot(ENABLED))[ENABLED] is False + null_path = "/raw-null" + events = raw.watch(null_path) + try: + await client.publish( + f"{TARGET}/settings{null_path}", + payload=b"null", + qos=1, + retain=True, + properties={"user_property": [("auth", "")]}, + ) + event = await wait_event_value(events, null_path, None) + assert event.retained + assert event.rev is None + await client.publish( + f"{TARGET}/settings{null_path}", + payload=b"", + qos=1, + retain=True, + properties={"user_property": [("auth", "")]}, + ) + event = await wait_event_delete(events, null_path) + assert event.retained + finally: + await events.aclose() await raw.set(ENABLED, True) assert settings.wait_payload(3.0, f"{TARGET}/settings{ENABLED}") == "true" assert await raw.get(ENABLED) is True From 043c1f5e617dad85f25db8b5678611afe19050b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Wed, 10 Jun 2026 09:26:49 +0200 Subject: [PATCH 3/4] web: clean up tree --- web/src/App.svelte | 5 ++- web/src/BrowseView.svelte | 30 ++++++----------- web/src/DiscoveryView.svelte | 51 +++++++++++------------------ web/src/TreeItem.svelte | 6 ++-- web/src/TreeRow.svelte | 14 ++++++-- web/src/lib/browse-model.ts | 2 +- web/src/lib/discovery-tree.ts | 16 ++++++--- web/src/lib/tree-interaction.ts | 5 +-- web/src/lib/tree-navigation.test.ts | 44 ++++++++++++++++--------- web/src/lib/tree-navigation.ts | 14 +++----- web/src/lib/tree-state.test.ts | 2 +- web/src/lib/tree-state.ts | 18 ++++++---- web/src/lib/tree-view.ts | 2 +- 13 files changed, 109 insertions(+), 100 deletions(-) diff --git a/web/src/App.svelte b/web/src/App.svelte index f19a5244..8169c0fc 100644 --- a/web/src/App.svelte +++ b/web/src/App.svelte @@ -121,11 +121,11 @@ } } - function navigateBrowseTree(path: string, direction: NavDirection, step?: number) { + function navigateBrowseTree(path: string, direction: NavDirection, step?: number): string { const next = browse.navigate(path, direction, step); browse.loadSelected(next); browse = browse; - focusTreeItem(next); + return next; } function commitSettings({ settings: nextSettings, changed }: SettingsCommit) { @@ -395,7 +395,6 @@ {settingsRevision} {status} {error} - rootNode={browse.rootNode} treeNodes={browse.treeNodes} selectedPath={browse.selectedPath} selected={browse.selected} diff --git a/web/src/BrowseView.svelte b/web/src/BrowseView.svelte index 53a79611..dd9f3cfd 100644 --- a/web/src/BrowseView.svelte +++ b/web/src/BrowseView.svelte @@ -3,7 +3,7 @@ import type { TreeActions, TreeNodeView } from "./lib/tree-view"; import SelectedPanel from "./SelectedPanel.svelte"; import StatusLog from "./StatusLog.svelte"; - import TreeItem from "./TreeItem.svelte"; + import TreeView from "./TreeView.svelte"; export let activePrefix: string; export let discoverHref: string; @@ -12,7 +12,6 @@ export let settingsRevision: string; export let status: string; export let error: string; - export let rootNode: ViewNode | undefined; export let treeNodes: Map; export let selectedPath: string; export let selected: ViewNode | undefined; @@ -51,17 +50,15 @@ {/if}
- {#if rootNode} -
    - -
+ {#if treeNodes.has(treeRoot)} + {:else}

No schema loaded.

{/if} @@ -113,13 +110,6 @@ min-width: 0; overflow: hidden; } - - .tree ul { - margin: 0; - min-width: 0; - padding: 0; - } - @media (min-width: 761px) { .browse { height: calc(100dvh - 2 * var(--space)); diff --git a/web/src/DiscoveryView.svelte b/web/src/DiscoveryView.svelte index e810edcd..5084af91 100644 --- a/web/src/DiscoveryView.svelte +++ b/web/src/DiscoveryView.svelte @@ -3,7 +3,7 @@ import { TreeInteraction } from "./lib/tree-interaction"; import type { TreeActions, TreeNodeView } from "./lib/tree-view"; import { type NavDirection } from "./lib/tree-navigation"; - import TreeItem from "./TreeItem.svelte"; + import TreeView from "./TreeView.svelte"; export let broker: string; export let discoveryPattern: string; @@ -18,7 +18,6 @@ $: nodes = discoveryTree(discoveredPrefixes); $: treeNodes = discoveryTreeView(nodes, browseHref); - $: rootNode = nodes.get(""); $: nextPrefixKey = discoveredPrefixes.map((prefix) => prefix.prefix).sort().join("\n"); $: if (nextPrefixKey !== prefixKey) { prefixKey = nextPrefixKey; @@ -46,23 +45,22 @@ interaction = interaction; } - function focusTreeItem(path: string) { - requestAnimationFrame(() => { - document - .querySelector(`[data-tree-path="${CSS.escape(path)}"]`) - ?.focus(); - }); - } - - function navigateTree(path: string, direction: NavDirection, step?: number) { - const next = interaction.navigate(visiblePaths, path, direction, step); + function navigateTree(path: string, direction: NavDirection, step?: number): string { + const next = interaction.navigate("", flatNodes, path, direction, step); interaction = interaction; - focusTreeItem(next); + return next; } $: treeActions = { + activate: (node: TreeNodeView, internal: boolean, open: boolean) => { + if (node.href) { + location.href = node.href; + } else if (internal) { + setExpanded(node.path, !open); + } + }, key: (node: TreeNodeView, direction: NavDirection, step?: number) => { - navigateTree(node.path, direction, step); + return navigateTree(node.path, direction, step); }, open: setExpanded, select, @@ -95,23 +93,12 @@ {#if discoveredPrefixes.length}

Prefixes

-
    - {#if rootNode} - - {/if} -
+
{/if} - - diff --git a/web/src/TreeItem.svelte b/web/src/TreeItem.svelte index 1b2a54b2..451078bd 100644 --- a/web/src/TreeItem.svelte +++ b/web/src/TreeItem.svelte @@ -17,7 +17,9 @@ $: open = expanded.has(node.path); $: flashedRow = flashed.has(node.path); $: title = node.href - ? "Enter opens. Ctrl-click opens a new tab. Arrows/Home/End/Page navigate." + ? internal + ? "Enter opens. Space toggles. Arrows/Home/End/Page navigate." + : "Enter opens. Ctrl-click opens a new tab. Arrows/Home/End/Page navigate." : internal ? "Enter or Space toggles. Arrows/Home/End/Page navigate." : "Enter edits. Arrows/Home/End/Page navigate."; @@ -37,7 +39,7 @@ // Enter activates, and focus stays on the selected row. switch (event.key) { case "Enter": - if (!node.href) { + if (!node.href || internal) { event.preventDefault(); if (actions.activate) { actions.activate(node, internal, open); diff --git a/web/src/TreeRow.svelte b/web/src/TreeRow.svelte index c8124e65..c2391ac8 100644 --- a/web/src/TreeRow.svelte +++ b/web/src/TreeRow.svelte @@ -43,7 +43,7 @@ } -{#if href} +{#if href && !internal} {label} + {:else} + {label} + {/if} {#if value} = {value} @@ -167,6 +171,12 @@ white-space: nowrap; } + a.label { + color: inherit; + text-decoration-thickness: 1px; + text-underline-offset: 0.15em; + } + .value { flex: 1 1 auto; min-width: 0; diff --git a/web/src/lib/browse-model.ts b/web/src/lib/browse-model.ts index 078e691a..cdb10503 100644 --- a/web/src/lib/browse-model.ts +++ b/web/src/lib/browse-model.ts @@ -108,7 +108,7 @@ export class BrowseModel { } navigate(path: string, direction: NavDirection, step?: number): string { - return this.interaction.navigate(this.visiblePaths, path, direction, step); + return this.interaction.navigate(this.root, this.tree.flatNodes, path, direction, step); } updateEditor(value: string): void { diff --git a/web/src/lib/discovery-tree.ts b/web/src/lib/discovery-tree.ts index 60029e45..21a3fe40 100644 --- a/web/src/lib/discovery-tree.ts +++ b/web/src/lib/discovery-tree.ts @@ -8,6 +8,7 @@ type PrefixEntry = { export type DiscoveryNode = { path: string; label: string; + parent?: string; prefix?: string; children: string[]; }; @@ -17,10 +18,10 @@ export function discoveryTree(prefixes: PrefixEntry[]): Map): Map { - return new Map([...nodes].map(([path, node]) => [path, { path, children: node.children }])); + return new Map([...nodes].map(([path, node]) => [ + path, + { + path, + ...(node.parent === undefined ? {} : { parent: node.parent }), + children: node.children, + }, + ])); } export function discoveryTreeView( diff --git a/web/src/lib/tree-interaction.ts b/web/src/lib/tree-interaction.ts index 33663f72..d2d21641 100644 --- a/web/src/lib/tree-interaction.ts +++ b/web/src/lib/tree-interaction.ts @@ -44,8 +44,9 @@ export class TreeInteraction { return visibleTreePaths(root, nodes, this.expanded); } - navigate(visible: string[], path: string, direction: NavDirection, step?: number): string { - const next = movePath(visible, path, direction, step); + navigate(root: string, nodes: Map, path: string, direction: NavDirection, step?: number): string { + const visible = this.visiblePaths(root, nodes); + const next = movePath(visible, path, direction, nodes, step); this.select(next); return next; } diff --git a/web/src/lib/tree-navigation.test.ts b/web/src/lib/tree-navigation.test.ts index 632a7d63..f9791425 100644 --- a/web/src/lib/tree-navigation.test.ts +++ b/web/src/lib/tree-navigation.test.ts @@ -6,25 +6,39 @@ describe("tree navigation", () => { it("walks visible rows in display order", () => { const nodes = new Map([ ["", { path: "", children: ["/a", "/b"] }], - ["/a", { path: "/a", children: ["/a/x"] }], - ["/a/x", { path: "/a/x", children: [] }], - ["/b", { path: "/b", children: [] }], + ["/a", { path: "/a", parent: "", children: ["/a/x"] }], + ["/a/x", { path: "/a/x", parent: "/a", children: [] }], + ["/b", { path: "/b", parent: "", children: [] }], ]); const visible = visibleTreePaths("", nodes, new Set(["", "/a"])); expect(visible).toEqual(["", "/a", "/a/x", "/b"]); - expect(movePath(visible, "/a", "next")).toBe("/a/x"); - expect(movePath(visible, "/a", "previous")).toBe(""); - expect(movePath(visible, "/a", "first")).toBe(""); - expect(movePath(visible, "/a", "last")).toBe("/b"); - expect(movePath(visible, "/a", "child")).toBe("/a/x"); - expect(movePath(visible, "/a/x", "parent")).toBe("/a"); - expect(movePath(visible, "", "parent")).toBe(""); - expect(movePath(["/a", "/a/x"], "/a", "parent")).toBe("/a"); - expect(movePath(visible, "", "pageNext", 2)).toBe("/a/x"); - expect(movePath(visible, "/b", "pagePrevious", 2)).toBe("/a"); - expect(movePath(visible, "/a", "pagePrevious", 20)).toBe(""); - expect(movePath(visible, "/a", "pageNext", 20)).toBe("/b"); + expect(movePath(visible, "/a", "next", nodes)).toBe("/a/x"); + expect(movePath(visible, "/a", "previous", nodes)).toBe(""); + expect(movePath(visible, "/a", "first", nodes)).toBe(""); + expect(movePath(visible, "/a", "last", nodes)).toBe("/b"); + expect(movePath(visible, "/a", "child", nodes)).toBe("/a/x"); + expect(movePath(visible, "/a/x", "parent", nodes)).toBe("/a"); + expect(movePath(visible, "", "parent", nodes)).toBe(""); + expect(movePath(["/a", "/a/x"], "/a", "parent", nodes)).toBe("/a"); + expect(movePath(visible, "", "pageNext", nodes, 2)).toBe("/a/x"); + expect(movePath(visible, "/b", "pagePrevious", nodes, 2)).toBe("/a"); + expect(movePath(visible, "/a", "pagePrevious", nodes, 20)).toBe(""); + expect(movePath(visible, "/a", "pageNext", nodes, 20)).toBe("/b"); + }); + + it("navigates by explicit structure rather than path syntax", () => { + const nodes = new Map([ + ["root", { path: "root", children: ["left", "right"] }], + ["left", { path: "left", parent: "root", children: ["leaf"] }], + ["leaf", { path: "leaf", parent: "left", children: [] }], + ["right", { path: "right", parent: "root", children: [] }], + ]); + const visible = visibleTreePaths("root", nodes, new Set(["root", "left"])); + + expect(visible).toEqual(["root", "left", "leaf", "right"]); + expect(movePath(visible, "left", "child", nodes)).toBe("leaf"); + expect(movePath(visible, "leaf", "parent", nodes)).toBe("left"); }); it("builds a browsable discovery tree from prefixes", () => { diff --git a/web/src/lib/tree-navigation.ts b/web/src/lib/tree-navigation.ts index a54039cb..649cb7e9 100644 --- a/web/src/lib/tree-navigation.ts +++ b/web/src/lib/tree-navigation.ts @@ -10,6 +10,7 @@ export type NavDirection = export type FlatTreeNode = { path: string; + parent?: string; children: string[]; }; @@ -41,6 +42,7 @@ export function movePath( visible: string[], current: string, direction: NavDirection, + nodes: Map, step = 1, ): string { if (!visible.length) { @@ -50,7 +52,7 @@ export function movePath( switch (direction) { case "child": { const child = visible[index + 1]; - return child && parentTreePath(child) === current ? child : current; + return child && nodes.get(child)?.parent === current ? child : current; } case "first": return visible[0]; @@ -65,7 +67,7 @@ export function movePath( case "previous": return visible[Math.max(index - 1, 0)]; case "parent": { - const parent = parentTreePath(current); + const parent = nodes.get(current)?.parent; return parent !== undefined && visible.includes(parent) ? parent : current; } } @@ -88,11 +90,3 @@ export function toggleExpansion( } return { expanded: nextExpanded, userClosed: nextUserClosed }; } - -function parentTreePath(path: string): string | undefined { - if (!path) { - return undefined; - } - const index = path.lastIndexOf("/"); - return index <= 0 ? "" : path.slice(0, index); -} diff --git a/web/src/lib/tree-state.test.ts b/web/src/lib/tree-state.test.ts index cbdf4499..b355581c 100644 --- a/web/src/lib/tree-state.test.ts +++ b/web/src/lib/tree-state.test.ts @@ -64,7 +64,7 @@ describe("tree state", () => { { path: "/a", kind: "leaf", children: [], present: true, value: 1 }, ]).values()]).toEqual([ { path: "", children: ["/a"] }, - { path: "/a", children: [] }, + { path: "/a", parent: "", children: [] }, ]); }); diff --git a/web/src/lib/tree-state.ts b/web/src/lib/tree-state.ts index 70da0a8c..93078caa 100644 --- a/web/src/lib/tree-state.ts +++ b/web/src/lib/tree-state.ts @@ -70,13 +70,17 @@ export function childrenByParent(nodes: ViewNode[]): Map { export function flatTreeNodes(nodes: ViewNode[]): Map { const children = childrenByParent(nodes); - return new Map(nodes.map((node) => [ - node.path, - { - path: node.path, - children: (children.get(node.path) ?? []).map((child) => child.path), - }, - ])); + return new Map(nodes.map((node) => { + const parent = parentPath(node.path); + return [ + node.path, + { + path: node.path, + ...(parent === undefined ? {} : { parent }), + children: (children.get(node.path) ?? []).map((child) => child.path), + }, + ]; + })); } export function treeViewNodes( diff --git a/web/src/lib/tree-view.ts b/web/src/lib/tree-view.ts index 133f2e70..78c63a51 100644 --- a/web/src/lib/tree-view.ts +++ b/web/src/lib/tree-view.ts @@ -12,7 +12,7 @@ export type TreeNodeView = { export type TreeActions = { activate?: (node: TreeNodeView, internal: boolean, open: boolean) => void; - key: (node: TreeNodeView, direction: NavDirection, step?: number) => void; + key: (node: TreeNodeView, direction: NavDirection, step?: number) => string; open: (path: string, open: boolean) => void; select: (path: string) => void; }; From cba3b0e3f4046316e06b5da54d2c202ea44a94ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Wed, 10 Jun 2026 09:36:06 +0200 Subject: [PATCH 4/4] coap: use &'static Schema --- miniconf_coap/examples/coap_server.rs | 3 ++- miniconf_coap/src/handler.rs | 22 +++++++++------------- miniconf_coap/src/lib.rs | 20 ++++++++++---------- miniconf_coap/src/schema.rs | 20 +++++++++----------- miniconf_coap/tests/packet_routes.rs | 11 ++++++++--- 5 files changed, 38 insertions(+), 38 deletions(-) diff --git a/miniconf_coap/examples/coap_server.rs b/miniconf_coap/examples/coap_server.rs index 0d2dfa07..62b29e60 100644 --- a/miniconf_coap/examples/coap_server.rs +++ b/miniconf_coap/examples/coap_server.rs @@ -7,6 +7,7 @@ use coap_message::error::RenderableOnMinimal as _; use coap_message_implementations::{inmemory, inmemory_write}; use coap_numbers::code; use defmt::{debug, info, warn}; +use miniconf::TreeSchema; use miniconf_coap::{Json, MiniconfCoapHandler, SchemaCoapHandler}; const DEFAULT_BIND: &str = "127.0.0.1:56830"; @@ -87,7 +88,7 @@ fn demo_handler() -> impl coap_handler::Handler + coap_handler::Reporting { new_dispatcher() .below(&["settings"], miniconf) - .below(&["schema"], SchemaCoapHandler::::json()) + .below(&["schema"], SchemaCoapHandler::json(Settings::SCHEMA)) .at( &["status"], SimpleRendered::new_typed_str(r#"{"ok":true}"#, Some(JSON_CONTENT_FORMAT)), diff --git a/miniconf_coap/src/handler.rs b/miniconf_coap/src/handler.rs index fb320fef..ef00c0d7 100644 --- a/miniconf_coap/src/handler.rs +++ b/miniconf_coap/src/handler.rs @@ -196,21 +196,20 @@ where /// #[cfg(feature = "json-core")] #[derive(Debug)] -pub struct SchemaCoapHandler(PhantomData); +pub struct SchemaCoapHandler { + schema: &'static Schema, +} #[cfg(feature = "json-core")] -impl SchemaCoapHandler { +impl SchemaCoapHandler { /// Create a route-relative JSON schema handler. - pub const fn json() -> Self { - Self(PhantomData) + pub const fn json(schema: &'static Schema) -> Self { + Self { schema } } } #[cfg(feature = "json-core")] -impl coap_handler::Handler for SchemaCoapHandler -where - Settings: TreeSchema, -{ +impl coap_handler::Handler for SchemaCoapHandler { type RequestData = CoapHandlerRequest; type ExtractRequestError = Error; type BuildResponseError = M::UnionError; @@ -234,7 +233,7 @@ where ) -> Result<(), Self::BuildResponseError> { let request = request.into_request_parts(); let mut response_buf = [0; MAX_HANDLER_RESPONSE_LENGTH]; - let outcome = SchemaRoute::new("").handle::(&request, &mut response_buf); + let outcome = SchemaRoute::new("", self.schema).handle(&request, &mut response_buf); let response = outcome.response().unwrap_or(Response { code: code::NOT_FOUND, content_format: None, @@ -289,10 +288,7 @@ const fn coap_uint_len(value: u16) -> usize { } #[cfg(feature = "json-core")] -impl coap_handler::Reporting for SchemaCoapHandler -where - Settings: TreeSchema, -{ +impl coap_handler::Reporting for SchemaCoapHandler { type Record<'res> = SchemaRecord where diff --git a/miniconf_coap/src/lib.rs b/miniconf_coap/src/lib.rs index 40321da0..3df065b6 100644 --- a/miniconf_coap/src/lib.rs +++ b/miniconf_coap/src/lib.rs @@ -290,13 +290,13 @@ mod tests { settings: &mut Settings, response_buf: &'a mut [u8], ) -> Outcome<'a> { - let schema = SchemaRoute::new("/schema"); + let schema = SchemaRoute::new("/schema", ::SCHEMA); let values = JsonValueRoute::json("/settings"); match request.path() { - "/schema" => return schema.handle::(request, response_buf), + "/schema" => return schema.handle(request, response_buf), path if path.starts_with("/schema/") => { - return schema.handle::(request, response_buf); + return schema.handle(request, response_buf); } "/settings" => return values.handle(request, settings, response_buf), path if path.starts_with("/settings/") => { @@ -399,10 +399,10 @@ mod tests { #[test] fn schema_route_is_separate() { init_host_logging(); - let handler = SchemaRoute::new("/schema"); + let handler = SchemaRoute::new("/schema", ::SCHEMA); let mut response = [0; 512]; let req = request(code::GET, &["schema"], None, b""); - let out = handler.handle::(&req, &mut response); + let out = handler.handle(&req, &mut response); let response = out.response().unwrap(); assert_eq!(response.code, code::CONTENT); assert_eq!( @@ -420,10 +420,10 @@ mod tests { #[test] fn schema_route_is_paged() { init_host_logging(); - let handler = SchemaRoute::new("/schema"); + let handler = SchemaRoute::new("/schema", ::SCHEMA); let mut response = [0; 512]; let req = request(code::GET, &["schema", "0"], None, b""); - let out = handler.handle::(&req, &mut response); + let out = handler.handle(&req, &mut response); let response = out.response().unwrap(); assert_eq!(response.code, code::CONTENT); assert_eq!( @@ -434,7 +434,7 @@ mod tests { let mut response = [0; 512]; let req = request(code::GET, &["schema", "99"], None, b""); - let out = handler.handle::(&req, &mut response); + let out = handler.handle(&req, &mut response); let response = out.response().unwrap(); assert_eq!(response.code, code::NOT_FOUND); assert_eq!(response.payload, br#"{"kind":"not_found","depth":1}"#); @@ -505,9 +505,9 @@ mod tests { content_format::from_str("application/json").unwrap(), ); let request = RequestParts::from_message(&request).unwrap(); - let handler = SchemaRoute::new("/schema"); + let handler = SchemaRoute::new("/schema", ::SCHEMA); let mut response = [0; 512]; - let outcome = handler.handle::(&request, &mut response); + let outcome = handler.handle(&request, &mut response); let response = outcome.response().unwrap(); assert_eq!(response.code, code::CONTENT); assert_eq!( diff --git a/miniconf_coap/src/schema.rs b/miniconf_coap/src/schema.rs index 87c6c76e..0df9245c 100644 --- a/miniconf_coap/src/schema.rs +++ b/miniconf_coap/src/schema.rs @@ -1,7 +1,7 @@ use coap_numbers::code; use defmt::{debug, trace, warn}; use miniconf::{ - TreeSchema, + Schema, compact_schema::{SchemaDefs, serialize_schema_page}, }; use serde::Serialize; @@ -11,10 +11,11 @@ use crate::{Error, MAX_SCHEMA_DEFS, Outcome, Problem, RequestParts, Response, fo const SCHEMA_PROTO: u8 = 1; -/// Schema route backed by `TreeSchema`. -#[derive(defmt::Format, Debug, Clone, Copy)] +/// Schema route backed by a Miniconf schema. +#[derive(Debug, Clone, Copy)] pub struct SchemaRoute<'a> { base: &'a str, + schema: &'static Schema, } impl<'a> SchemaRoute<'a> { @@ -22,19 +23,16 @@ impl<'a> SchemaRoute<'a> { /// /// The base path serves a JSON manifest. `base/{page}` serves newline-delimited compact schema /// pages. - pub const fn new(base: &'a str) -> Self { - Self { base } + pub const fn new(base: &'a str, schema: &'static Schema) -> Self { + Self { base, schema } } /// Handle a schema `GET` request. - pub fn handle<'b, Settings>( + pub fn handle<'b>( &self, request: &RequestParts<'_>, response_buf: &'b mut [u8], - ) -> Outcome<'b> - where - Settings: TreeSchema, - { + ) -> Outcome<'b> { let Some(resource) = self.resource(request.path()) else { trace!("Ignoring non-schema CoAP route request={}", request); return Outcome::Unhandled; @@ -49,7 +47,7 @@ impl<'a> SchemaRoute<'a> { .response(response_buf), ); } - let Ok(defs) = SchemaDefs::::new(Settings::SCHEMA) else { + let Ok(defs) = SchemaDefs::::new(self.schema) else { return Outcome::Handled( Error::new(code::INTERNAL_SERVER_ERROR, Problem::Serialization) .response(response_buf), diff --git a/miniconf_coap/tests/packet_routes.rs b/miniconf_coap/tests/packet_routes.rs index a0092db8..b7207b25 100644 --- a/miniconf_coap/tests/packet_routes.rs +++ b/miniconf_coap/tests/packet_routes.rs @@ -154,9 +154,11 @@ mod json { response_buf: &'a mut [u8], ) -> Outcome<'a> { match request.path() { - "/schema" => SchemaRoute::new("/schema").handle::(request, response_buf), + "/schema" => SchemaRoute::new("/schema", ::SCHEMA) + .handle(request, response_buf), path if path.starts_with("/schema/") => { - SchemaRoute::new("/schema").handle::(request, response_buf) + SchemaRoute::new("/schema", ::SCHEMA) + .handle(request, response_buf) } "/settings" => settings_route(request, settings, response_buf), path if path.starts_with("/settings/") => { @@ -500,7 +502,10 @@ mod coap_handler { &["settings"], MiniconfCoapHandler::::json(Settings::default()), ) - .below(&["schema"], SchemaCoapHandler::::json()) + .below( + &["schema"], + SchemaCoapHandler::json(::SCHEMA), + ) .at( &["status"], SimpleRendered::new_typed_str(