gmqtt misc

CHANGELOG.md

@@ -33,8 +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 them to
-  invalidate cached schema and settings.
+* 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.
 

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::<Settings>::json())
+        .below(&["schema"], SchemaCoapHandler::json(Settings::SCHEMA))
         .at(
             &["status"],
             SimpleRendered::new_typed_str(r#"{"ok":true}"#, Some(JSON_CONTENT_FORMAT)),

miniconf_coap/src/handler.rs

@@ -196,21 +196,20 @@ where
 ///
 #[cfg(feature = "json-core")]
 #[derive(Debug)]
-pub struct SchemaCoapHandler<Settings>(PhantomData<Settings>);
+pub struct SchemaCoapHandler {
+    schema: &'static Schema,
+}
 
 #[cfg(feature = "json-core")]
-impl<Settings> SchemaCoapHandler<Settings> {
+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<Settings> coap_handler::Handler for SchemaCoapHandler<Settings>
-where
-    Settings: TreeSchema,
-{
+impl coap_handler::Handler for SchemaCoapHandler {
     type RequestData = CoapHandlerRequest;
     type ExtractRequestError = Error;
     type BuildResponseError<M: MinimalWritableMessage> = M::UnionError;
@@ -234,7 +233,7 @@ where
     ) -> Result<(), Self::BuildResponseError<M>> {
         let request = request.into_request_parts();
         let mut response_buf = [0; MAX_HANDLER_RESPONSE_LENGTH];
-        let outcome = SchemaRoute::new("").handle::<Settings>(&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<Settings> coap_handler::Reporting for SchemaCoapHandler<Settings>
-where
-    Settings: TreeSchema,
-{
+impl coap_handler::Reporting for SchemaCoapHandler {
     type Record<'res>
         = SchemaRecord
     where

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", <Settings as miniconf::TreeSchema>::SCHEMA);
         let values = JsonValueRoute::json("/settings");
 
         match request.path() {
-            "/schema" => return schema.handle::<Settings>(request, response_buf),
+            "/schema" => return schema.handle(request, response_buf),
             path if path.starts_with("/schema/") => {
-                return schema.handle::<Settings>(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", <Settings as miniconf::TreeSchema>::SCHEMA);
         let mut response = [0; 512];
         let req = request(code::GET, &["schema"], None, b"");
-        let out = handler.handle::<Settings>(&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", <Settings as miniconf::TreeSchema>::SCHEMA);
         let mut response = [0; 512];
         let req = request(code::GET, &["schema", "0"], None, b"");
-        let out = handler.handle::<Settings>(&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::<Settings>(&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", <Settings as miniconf::TreeSchema>::SCHEMA);
         let mut response = [0; 512];
-        let outcome = handler.handle::<Settings>(&request, &mut response);
+        let outcome = handler.handle(&request, &mut response);
         let response = outcome.response().unwrap();
         assert_eq!(response.code, code::CONTENT);
         assert_eq!(

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,30 +11,28 @@ 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> {
     /// Construct a compact schema route.
     ///
     /// 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::<MAX_SCHEMA_DEFS>::new(Settings::SCHEMA) else {
+        let Ok(defs) = SchemaDefs::<MAX_SCHEMA_DEFS>::new(self.schema) else {
             return Outcome::Handled(
                 Error::new(code::INTERNAL_SERVER_ERROR, Problem::Serialization)
                     .response(response_buf),

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::<Settings>(request, response_buf),
+            "/schema" => SchemaRoute::new("/schema", <Settings as miniconf::TreeSchema>::SCHEMA)
+                .handle(request, response_buf),
             path if path.starts_with("/schema/") => {
-                SchemaRoute::new("/schema").handle::<Settings>(request, response_buf)
+                SchemaRoute::new("/schema", <Settings as miniconf::TreeSchema>::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::<Settings, Json>::json(Settings::default()),
             )
-            .below(&["schema"], SchemaCoapHandler::<Settings>::json())
+            .below(
+                &["schema"],
+                SchemaCoapHandler::json(<Settings as miniconf::TreeSchema>::SCHEMA),
+            )
             .at(
                 &["status"],
                 SimpleRendered::new_typed_str(

py/README.md

@@ -11,29 +11,28 @@ 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()
-        await mc.set("/path", 42)
+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 with mc.track("/subtree") as tracked:
-            await tracked.wait_ready()
-            value = tracked.cached("/subtree/leaf")
-            cached = tracked.snapshot()
+    async for event in mc.watch("/subtree"):
+        print(event.path, event.value if event.present else "<deleted>")
 ```
 
 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:
 
@@ -46,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: