diff --git a/Cargo.lock b/Cargo.lock index d89b2abe..d04991b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1829,6 +1829,7 @@ name = "embassy-mqtt-connector-demo" version = "0.1.0" dependencies = [ "aimdb-core", + "aimdb-data-contracts", "aimdb-embassy-adapter", "aimdb-mqtt-connector", "aimdb-serial-connector", @@ -3586,7 +3587,6 @@ dependencies = [ "aimdb-data-contracts", "defmt 1.1.1", "serde", - "serde_json", ] [[package]] diff --git a/examples/embassy-mqtt-connector-demo/Cargo.toml b/examples/embassy-mqtt-connector-demo/Cargo.toml index fc1b22af..34ba6fbb 100644 --- a/examples/embassy-mqtt-connector-demo/Cargo.toml +++ b/examples/embassy-mqtt-connector-demo/Cargo.toml @@ -40,10 +40,10 @@ aimdb-serial-connector = { path = "../../aimdb-serial-connector", default-featur # Shared demo logic. `serde` makes the shared types serde-codec'able in no_std so # they can be exposed over the serial remote-access server. -mqtt-connector-demo-common = { path = "../mqtt-connector-demo-common", features = [ - "alloc", - "derive", - "serde", +mqtt-connector-demo-common = { path = "../mqtt-connector-demo-common" } +# publish / subscribe; the common crate enables linkable-json for its contracts. +aimdb-data-contracts = { path = "../../aimdb-data-contracts", default-features = false, features = [ + "linkable", ] } # Embassy ecosystem - STM32H563ZI with Ethernet diff --git a/examples/embassy-mqtt-connector-demo/src/main.rs b/examples/embassy-mqtt-connector-demo/src/main.rs index 78c52bdb..5b6277ec 100644 --- a/examples/embassy-mqtt-connector-demo/src/main.rs +++ b/examples/embassy-mqtt-connector-demo/src/main.rs @@ -69,7 +69,8 @@ extern crate alloc; use aimdb_core::remote::SecurityPolicy; -use aimdb_core::{AimDbBuilder, Producer, RecordKey, RuntimeContext}; +use aimdb_core::{AimDbBuilder, RecordKey}; +use aimdb_data_contracts::{publish, subscribe}; use aimdb_embassy_adapter::io::EmbassyUart; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; use aimdb_serial_connector::SerialServer; @@ -91,9 +92,10 @@ use aimdb_mqtt_connector::MqttConnector; #[cfg(feature = "tls")] use aimdb_mqtt_connector::TlsOptions; -// Import shared types, monitors, and compile-time safe keys from the common crate +// Shared contracts, keys, thermometers and monitors from the common crate use mqtt_connector_demo_common::{ - CommandKey, SensorKey, Temperature, TemperatureCommand, command_consumer, temperature_logger, + CommandKey, SensorKey, SimThermometers, Temperature, TemperatureCommand, Thermometer, + command_consumer, temperature_logger, }; // Simple embedded allocator (required by some dependencies) @@ -116,71 +118,6 @@ async fn net_task(mut runner: embassy_net::Runner<'static, Device>) -> ! { runner.run().await } -// ============================================================================ -// TEMPERATURE PRODUCERS (platform-specific due to embassy-time) -// -// Each cycles its readings endlessly rather than stopping after a fixed -// count: reconnect, re-subscribe and ping cadence only become observable -// while something is still publishing. -// ============================================================================ - -/// Indoor temperature sensor producer -async fn indoor_temp_producer(ctx: RuntimeContext, temperature: Producer) { - let log = ctx.log(); - log.info("🏠 Starting INDOOR temperature producer...\n"); - - for i in (0..5).cycle() { - let temp = Temperature::new("indoor-001", 22.0 + (i as f32 * 0.5)); // Indoor temps: 22-24Β°C - - log.info(&alloc::format!( - "🏠 Indoor sensor producing: {:.1}Β°C", - temp.celsius - )); - - temperature.produce(temp); - - Timer::after(Duration::from_secs(2)).await; - } -} - -/// Outdoor temperature sensor producer -async fn outdoor_temp_producer(ctx: RuntimeContext, temperature: Producer) { - let log = ctx.log(); - log.info("🌳 Starting OUTDOOR temperature producer...\n"); - - for i in (0..5).cycle() { - let temp = Temperature::new("outdoor-001", 5.0 + (i as f32 * 1.0)); // Outdoor temps: 5-9Β°C (cold!) - - log.info(&alloc::format!( - "🌳 Outdoor sensor producing: {:.1}Β°C", - temp.celsius - )); - - temperature.produce(temp); - - Timer::after(Duration::from_secs(2)).await; - } -} - -/// Server room temperature sensor producer -async fn server_room_temp_producer(ctx: RuntimeContext, temperature: Producer) { - let log = ctx.log(); - log.info("πŸ–₯️ Starting SERVER ROOM temperature producer...\n"); - - for i in (0..5).cycle() { - let temp = Temperature::new("server-room-001", 18.0 + (i as f32 * 0.2)); // Server room: 18-19Β°C (cooled) - - log.info(&alloc::format!( - "πŸ–₯️ Server room sensor producing: {:.1}Β°C", - temp.celsius - )); - - temperature.produce(temp); - - Timer::after(Duration::from_secs(2)).await; - } -} - // ============================================================================ // MQTT CONFIGURATION // ============================================================================ @@ -428,13 +365,21 @@ async fn main(spawner: Spawner) { .with_keep_alive(MQTT_KEEP_ALIVE) }; - let mut builder = AimDbBuilder::new() - .runtime(runtime.clone()) - .with_connector(mqtt) - .with_connector( - SerialServer::new(EmbassyUart::new(serial_rx, serial_tx)) - .security_policy(SecurityPolicy::read_only()), - ); + // The serial server takes no links, so it needs no handle. + let mut builder = AimDbBuilder::new().runtime(runtime.clone()).with_connector( + SerialServer::new(EmbassyUart::new(serial_rx, serial_tx)) + .security_policy(SecurityPolicy::read_only()), + ); + let mqtt = builder.connect(mqtt); + + // The simulated thermometers cycle their readings endlessly: reconnect, + // re-subscribe and ping cadence only become observable while something + // is still publishing. + let SimThermometers { + indoor, + outdoor, + server_room, + } = SimThermometers::new(); // ======================================================================== // TEMPERATURE SENSORS (outbound: AimDB β†’ MQTT) @@ -447,31 +392,25 @@ async fn main(spawner: Spawner) { builder.configure::(SensorKey::TempIndoor, |reg| { reg.buffer_sized::<16, 3>(EmbassyBufferType::SpmcRing) .with_remote_access() - .source(indoor_temp_producer) + .input(Thermometer::on(indoor)) .tap(temperature_logger) - .link_to(SensorKey::TempIndoor.link_address().unwrap()) - .with_serializer(|_ctx, temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .output(publish(mqtt, SensorKey::TempIndoor.link_address().unwrap())); }); builder.configure::(SensorKey::TempOutdoor, |reg| { reg.buffer_sized::<16, 3>(EmbassyBufferType::SpmcRing) .with_remote_access() - .source(outdoor_temp_producer) + .input(Thermometer::on(outdoor)) .tap(temperature_logger) - .link_to(SensorKey::TempOutdoor.link_address().unwrap()) - .with_serializer(|_ctx, temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .output(publish(mqtt, SensorKey::TempOutdoor.link_address().unwrap())); }); builder.configure::(SensorKey::TempServerRoom, |reg| { reg.buffer_sized::<16, 3>(EmbassyBufferType::SpmcRing) .with_remote_access() - .source(server_room_temp_producer) + .input(Thermometer::on(server_room)) .tap(temperature_logger) - .link_to(SensorKey::TempServerRoom.link_address().unwrap()) - .with_serializer(|_ctx, temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .output(publish(mqtt, SensorKey::TempServerRoom.link_address().unwrap())); }); // ======================================================================== @@ -483,18 +422,14 @@ async fn main(spawner: Spawner) { reg.buffer_sized::<8, 2>(EmbassyBufferType::SpmcRing) .with_remote_access() .tap(command_consumer) - .link_from(CommandKey::TempIndoor.link_address().unwrap()) - .with_deserializer(|_ctx, data: &[u8]| TemperatureCommand::from_json(data)) - .finish(); + .input(subscribe(mqtt, CommandKey::TempIndoor.link_address().unwrap())); }); builder.configure::(CommandKey::TempOutdoor, |reg| { reg.buffer_sized::<8, 2>(EmbassyBufferType::SpmcRing) .with_remote_access() .tap(command_consumer) - .link_from(CommandKey::TempOutdoor.link_address().unwrap()) - .with_deserializer(|_ctx, data: &[u8]| TemperatureCommand::from_json(data)) - .finish(); + .input(subscribe(mqtt, CommandKey::TempOutdoor.link_address().unwrap())); }); info!("βœ… Database configured with multi-sensor MQTT:"); diff --git a/examples/mqtt-connector-demo-common/Cargo.toml b/examples/mqtt-connector-demo-common/Cargo.toml index 89806803..7917eff4 100644 --- a/examples/mqtt-connector-demo-common/Cargo.toml +++ b/examples/mqtt-connector-demo-common/Cargo.toml @@ -8,33 +8,23 @@ description = "Shared business logic for MQTT demos - runs on MCU, edge, and clo [features] default = [] -std = ["alloc", "aimdb-core/std", "serde", "serde_json"] -alloc = ["aimdb-core/alloc"] -derive = ["aimdb-core/derive"] -# serde derive on the shared types β€” works in `no_std + alloc` too (no serde_json -# needed: the no_std JSON paths are hand-rolled), so an embassy demo can expose -# these records over a remote-access (serial) server. -serde = ["dep:serde"] -# `SchemaType` + JSON `#[derive(Linkable)]`. `no_std + alloc` -# compatible (`linkable-json` uses serde_json's alloc-only configuration). -# Not enabled by default: the -# embassy demo's hand-rolled no_std JSON path is an intentional illustration, -# left untouched; only the tokio demo opts into this feature. -data-contracts = ["alloc", "serde", "dep:aimdb-data-contracts"] - +std = ["aimdb-core/std", "aimdb-data-contracts/std", "serde/std"] # Embassy-specific features (no_std) defmt = ["dep:defmt"] [dependencies] -# Core AimDB - runtime agnostic -aimdb-core = { path = "../../aimdb-core", default-features = false } -# Serialization (std only) -serde = { workspace = true, features = ["derive"], optional = true } -serde_json = { workspace = true, optional = true } -# Optional: SchemaType + JSON #[derive(Linkable)] (no_std + alloc) -aimdb-data-contracts = { path = "../../aimdb-data-contracts", default-features = false, optional = true, features = [ +# Core AimDB - runtime agnostic; `derive` for the `RecordKey` keys +aimdb-core = { path = "../../aimdb-core", default-features = false, features = [ + "alloc", + "derive", +] } +# The contracts' wire format: JSON `#[derive(Linkable)]`, `no_std + alloc` +# compatible (serde_json's alloc-only configuration), for publish/subscribe +aimdb-data-contracts = { path = "../../aimdb-data-contracts", default-features = false, features = [ "linkable-json", ] } +# Serialize/Deserialize on the contracts: the wire format, and remote access +serde = { workspace = true } # Optional defmt for embedded logging defmt = { workspace = true, optional = true } diff --git a/examples/mqtt-connector-demo-common/README.md b/examples/mqtt-connector-demo-common/README.md index abe9822d..9af31d14 100644 --- a/examples/mqtt-connector-demo-common/README.md +++ b/examples/mqtt-connector-demo-common/README.md @@ -14,7 +14,7 @@ The same monitor functions, data types, and business logic work across all platf β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ mqtt-connector-demo-common β”‚ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ -β”‚ β”‚ Data Types β”‚ β”‚ Monitor Functionsβ”‚ β”‚ +β”‚ β”‚ Contracts β”‚ β”‚ Monitor Functionsβ”‚ β”‚ β”‚ β”‚ Temperature β”‚ β”‚ temperature_loggerβ”‚ β”‚ β”‚ β”‚ TemperatureCmd β”‚ β”‚ command_consumer β”‚ β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ @@ -31,39 +31,47 @@ The same monitor functions, data types, and business logic work across all platf β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` +## Contents + +- **Contracts** (`types`): `Temperature`, `TemperatureCommand`. Their wire format is JSON through `#[derive(Linkable)]` on every target, `no_std + alloc` included. +- **Keys** (`keys`): `SensorKey`, `CommandKey`, each with its MQTT topic in `#[link_address]`. +- **Simulated thermometers** (`sim`): `SimThermometers` hands out one owned channel per sensor; `Thermometer::on(channel)` is an input for every record type built `From`. +- **Monitors** (`monitors`): `temperature_logger`, `command_consumer`. + ## Features -- `std` - Enable std support (serde_json for serialization) -- `alloc` - Enable allocator support (required for most features) +- `std` - Enable std support - `defmt` - Enable defmt integration for embedded logging ## Usage +The wiring is the same on both runtimes; only the buffer differs. + ### Tokio (Edge/Cloud) ```rust -use mqtt_connector_demo_common::{Temperature, TemperatureCommand, temperature_logger, command_consumer}; +use aimdb_data_contracts::publish; +use mqtt_connector_demo_common::{SensorKey, SimThermometers, Temperature, Thermometer, temperature_logger}; -builder.configure::("sensor.temp.indoor", |reg| { +let mqtt = builder.connect(MqttConnector::new(broker_url)); +let SimThermometers { indoor, .. } = SimThermometers::new(); + +builder.configure::(SensorKey::TempIndoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .input(Thermometer::on(indoor)) .tap(temperature_logger) // Shared monitor - .link_to("mqtt://sensors/temp/indoor") - .with_serializer(|temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .output(publish(mqtt, SensorKey::TempIndoor.link_address().unwrap())); }); ``` ### Embassy (MCU) ```rust -use mqtt_connector_demo_common::{Temperature, TemperatureCommand, temperature_logger, command_consumer}; - -builder.configure::("sensor.temp.indoor", |reg| { +builder.configure::(SensorKey::TempIndoor, |reg| { reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing) - .tap(temperature_logger) // Same monitor works! - .link_to("mqtt://sensors/temp/indoor") - .with_serializer(|temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .input(Thermometer::on(indoor)) // Same input works! + .tap(temperature_logger) // Same monitor works! + .output(publish(mqtt, SensorKey::TempIndoor.link_address().unwrap())); }); ``` diff --git a/examples/mqtt-connector-demo-common/src/lib.rs b/examples/mqtt-connector-demo-common/src/lib.rs index 397d24a3..505bead7 100644 --- a/examples/mqtt-connector-demo-common/src/lib.rs +++ b/examples/mqtt-connector-demo-common/src/lib.rs @@ -9,27 +9,34 @@ //! //! This crate demonstrates AimDB's core strength: **write once, run anywhere**. //! -//! The same monitor functions, data types, and business logic work across all -//! platforms by being generic over the `RuntimeAdapter` trait. Only the -//! platform-specific initialization (hardware setup, network stack) differs. +//! Everything here is runtime-agnostic: the contracts and their wire format +//! ([`types`]), the record keys and their MQTT topics ([`keys`]), the +//! simulated thermometers as an input ([`sim`]), and the monitors +//! ([`monitors`]). A demo adds only its platform setup and the record wiring: //! -//! # Compile-Time Safe Keys +//! ```rust,ignore +//! let mqtt = builder.connect(MqttConnector::new(broker_url)); +//! let SimThermometers { indoor, .. } = SimThermometers::new(); //! -//! This crate also demonstrates the `RecordKey` derive macro for type-safe -//! record keys. See the `keys` module for examples β€” unlinked, as it exists -//! only behind the `derive` feature. +//! builder.configure::(SensorKey::TempIndoor, |reg| { +//! reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) +//! .input(Thermometer::on(indoor)) +//! .tap(temperature_logger) +//! .output(publish(mqtt, SensorKey::TempIndoor.link_address().unwrap())); +//! }); +//! ``` #![cfg_attr(not(feature = "std"), no_std)] extern crate alloc; -#[cfg(feature = "derive")] pub mod keys; pub mod monitors; +pub mod sim; pub mod types; // Re-export main types -#[cfg(feature = "derive")] pub use keys::{CommandKey, SensorKey}; pub use monitors::{command_consumer, temperature_logger}; +pub use sim::{Reading, SimChannel, SimThermometers, Thermometer}; pub use types::{Temperature, TemperatureCommand}; diff --git a/examples/mqtt-connector-demo-common/src/sim.rs b/examples/mqtt-connector-demo-common/src/sim.rs new file mode 100644 index 00000000..f614a420 --- /dev/null +++ b/examples/mqtt-connector-demo-common/src/sim.rs @@ -0,0 +1,148 @@ +//! Simulated thermometers: one input type for every runtime. +//! +//! [`SimThermometers`] stands in for a board: it hands out one owned +//! [`SimChannel`] per sensor. [`Thermometer::on`] takes a channel by value and +//! is an [`Input`] for every record type that can be built from a +//! [`Reading`], so the same input serves Tokio and Embassy alike: +//! +//! ```rust,ignore +//! let SimThermometers { indoor, outdoor, server_room } = SimThermometers::new(); +//! +//! builder.configure::(SensorKey::TempIndoor, |reg| { +//! reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) +//! .input(Thermometer::on(indoor)); +//! }); +//! ``` +//! +//! A channel is not `Clone`, so wiring one sensor into two records is a +//! compile error (`use of moved value`) rather than two producers reading +//! one device. + +use aimdb_core::{Input, RecordRegistrar}; +use core::fmt::Debug; + +use crate::types::Temperature; + +/// What a thermometer yields: one temperature reading. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Reading { + /// Which sensor took the reading (e.g. "indoor-001"). + pub sensor_id: &'static str, + /// Temperature in degrees Celsius. + pub celsius: f32, +} + +impl From for Temperature { + fn from(reading: Reading) -> Self { + Temperature::new(reading.sensor_id, reading.celsius) + } +} + +/// One simulated sensor. Deliberately neither `Clone` nor `Copy`: an input +/// takes its channel by value. +#[derive(Debug)] +pub struct SimChannel { + sensor_id: &'static str, + /// First reading. + base: f32, + /// Added per reading; the sensor cycles through five readings. + step: f32, +} + +/// The simulated thermometers, one field per sensor. +#[derive(Debug)] +pub struct SimThermometers { + /// Indoor: 22.0–24.0 Β°C. + pub indoor: SimChannel, + /// Outdoor: 5.0–9.0 Β°C. + pub outdoor: SimChannel, + /// Server room: 18.0–18.8 Β°C. + pub server_room: SimChannel, +} + +impl SimThermometers { + /// The demo's three sensors. + pub fn new() -> Self { + let channel = |sensor_id, base, step| SimChannel { + sensor_id, + base, + step, + }; + Self { + indoor: channel("indoor-001", 22.0, 0.5), + outdoor: channel("outdoor-001", 5.0, 1.0), + server_room: channel("server-room-001", 18.0, 0.2), + } + } +} + +impl Default for SimThermometers { + fn default() -> Self { + Self::new() + } +} + +/// A thermometer input on one [`SimChannel`]: produces a reading every two +/// seconds, forever unless limited with [`readings`](Self::readings). +#[derive(Debug)] +pub struct Thermometer { + channel: SimChannel, + period_secs: u64, + readings: Option, +} + +impl Thermometer { + /// A thermometer reading `channel`. + pub fn on(channel: SimChannel) -> Self { + Self { + channel, + period_secs: 2, + readings: None, + } + } + + /// Seconds between readings (default 2). + pub fn every_secs(mut self, secs: u64) -> Self { + self.period_secs = secs; + self + } + + /// Stops after `count` readings instead of running forever. + pub fn readings(mut self, count: usize) -> Self { + self.readings = Some(count); + self + } +} + +impl Input for Thermometer +where + T: From + Send + Sync + 'static + Debug + Clone, +{ + fn wire(self, reg: &mut RecordRegistrar<'_, T>) { + let Thermometer { + channel, + period_secs, + readings, + } = self; + reg.source(move |ctx, producer| async move { + let log = ctx.log(); + log.info(&alloc::format!("🌑️ {} started", channel.sensor_id)); + + for i in (0..5u8).cycle().take(readings.unwrap_or(usize::MAX)) { + let reading = Reading { + sensor_id: channel.sensor_id, + celsius: channel.base + f32::from(i) * channel.step, + }; + log.info(&alloc::format!( + "🌑️ {} producing: {:.1}Β°C", + reading.sensor_id, + reading.celsius + )); + producer.produce(T::from(reading)); + ctx.time().sleep_secs(period_secs).await; + } + + log.info(&alloc::format!("βœ… {} finished", channel.sensor_id)); + }); + } +} diff --git a/examples/mqtt-connector-demo-common/src/types.rs b/examples/mqtt-connector-demo-common/src/types.rs index 76df37bf..c9ac9f02 100644 --- a/examples/mqtt-connector-demo-common/src/types.rs +++ b/examples/mqtt-connector-demo-common/src/types.rs @@ -1,16 +1,16 @@ //! Shared Data Types for MQTT Demo //! -//! These types work in both `std` and `no_std` environments using `alloc::string::String`. +//! The demo's contracts. They work in both `std` and `no_std` environments +//! using `alloc::string::String`, and their wire format is JSON through +//! `#[derive(Linkable)]` on every target, so `publish` / `subscribe` and +//! remote access use the same encoding on Tokio and Embassy. extern crate alloc; use alloc::string::String; -#[cfg(feature = "serde")] -use serde::{Deserialize, Serialize}; - -#[cfg(feature = "data-contracts")] use aimdb_data_contracts::{Linkable, SchemaType}; +use serde::{Deserialize, Serialize}; // ============================================================================ // TEMPERATURE READING @@ -19,9 +19,7 @@ use aimdb_data_contracts::{Linkable, SchemaType}; /// Temperature reading from a sensor /// /// Represents a temperature measurement that can be published to MQTT. -#[derive(Clone, Debug)] -#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] -#[cfg_attr(feature = "data-contracts", derive(Linkable))] +#[derive(Clone, Debug, Serialize, Deserialize, Linkable)] pub struct Temperature { /// Sensor identifier (e.g., "indoor-001", "outdoor-001") pub sensor_id: String, @@ -29,7 +27,6 @@ pub struct Temperature { pub celsius: f32, } -#[cfg(feature = "data-contracts")] impl SchemaType for Temperature { const NAME: &'static str = "mqtt_demo.temperature"; } @@ -55,32 +52,6 @@ impl Temperature { "πŸ“Š" } } - - /// Serialize to JSON format (works on both std and no_std) - pub fn to_json_vec(&self) -> alloc::vec::Vec { - #[cfg(feature = "std")] - { - serde_json::to_vec(self).unwrap_or_default() - } - #[cfg(not(feature = "std"))] - { - use alloc::format; - // Manual float formatting with 1 decimal place for no_std - // This produces identical JSON to the serde path for interop - let whole = self.celsius as i32; - let frac = ((self.celsius - whole as f32).abs() * 10.0 + 0.5) as i32 % 10; - let sign = if self.celsius < 0.0 && whole == 0 { - "-" - } else { - "" - }; - format!( - r#"{{"sensor_id":"{}","celsius":{}{}.{}}}"#, - self.sensor_id, sign, whole, frac - ) - .into_bytes() - } - } } // ============================================================================ @@ -88,9 +59,7 @@ impl Temperature { // ============================================================================ /// Command for controlling temperature sensors (inbound from MQTT) -#[derive(Clone, Debug)] -#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] -#[cfg_attr(feature = "data-contracts", derive(Linkable))] +#[derive(Clone, Debug, Serialize, Deserialize, Linkable)] pub struct TemperatureCommand { /// Action to perform (e.g., "read", "calibrate", "reset") pub action: String, @@ -98,7 +67,6 @@ pub struct TemperatureCommand { pub sensor_id: String, } -#[cfg(feature = "data-contracts")] impl SchemaType for TemperatureCommand { const NAME: &'static str = "mqtt_demo.temperature_command"; } @@ -111,51 +79,4 @@ impl TemperatureCommand { sensor_id: sensor_id.into(), } } - - /// Parse from JSON bytes (works on both std and no_std) - pub fn from_json(data: &[u8]) -> Result { - #[cfg(feature = "std")] - { - serde_json::from_slice(data) - .map_err(|e| alloc::format!("Failed to deserialize command: {}", e)) - } - #[cfg(not(feature = "std"))] - { - Self::parse_json_no_std(data) - } - } - - /// Simple JSON parser for no_std environments - #[cfg(not(feature = "std"))] - fn parse_json_no_std(data: &[u8]) -> Result { - use alloc::string::ToString; - use alloc::vec::Vec; - - let text = core::str::from_utf8(data).map_err(|_| "Invalid UTF-8".to_string())?; - - let mut action = String::new(); - let mut sensor_id = String::new(); - - for pair in text.trim_matches(|c| c == '{' || c == '}').split(',') { - let parts: Vec<&str> = pair.split(':').collect(); - if parts.len() != 2 { - continue; - } - - let key = parts[0].trim().trim_matches('"'); - let value = parts[1].trim().trim_matches('"'); - - match key { - "action" => action = value.into(), - "sensor_id" => sensor_id = value.into(), - _ => {} - } - } - - if action.is_empty() || sensor_id.is_empty() { - return Err("Missing required fields".to_string()); - } - - Ok(TemperatureCommand { action, sensor_id }) - } } diff --git a/examples/tokio-mqtt-connector-demo/Cargo.toml b/examples/tokio-mqtt-connector-demo/Cargo.toml index 51e150d3..9db266ab 100644 --- a/examples/tokio-mqtt-connector-demo/Cargo.toml +++ b/examples/tokio-mqtt-connector-demo/Cargo.toml @@ -22,12 +22,9 @@ aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = [ # Shared demo logic mqtt-connector-demo-common = { path = "../mqtt-connector-demo-common", features = [ "std", - "derive", - "data-contracts", ] } -# Format-neutral LinkableRegistrarExt; the common crate enables linkable-json -# for its derived value types. +# publish / subscribe; the common crate enables linkable-json for its contracts. aimdb-data-contracts = { path = "../../aimdb-data-contracts", features = [ "linkable", ] } diff --git a/examples/tokio-mqtt-connector-demo/src/main.rs b/examples/tokio-mqtt-connector-demo/src/main.rs index 0434b444..4cce4e8e 100644 --- a/examples/tokio-mqtt-connector-demo/src/main.rs +++ b/examples/tokio-mqtt-connector-demo/src/main.rs @@ -4,8 +4,15 @@ //! - Multiple temperature sensors publishing to different topics //! - Multiple command consumers receiving from different topics //! -//! This demo uses `mqtt-connector-demo-common` for shared types and monitors, -//! demonstrating AimDB's "write once, run anywhere" capability. +//! This demo uses `mqtt-connector-demo-common` for the shared contracts, keys, +//! simulated thermometers and monitors, demonstrating AimDB's "write once, run +//! anywhere" capability: this file holds only the setup and the record wiring. +//! +//! ## Wiring +//! +//! The connector is registered with `connect`, which returns a handle. Records +//! take a thermometer as `.input(..)` and link to MQTT with `publish` / +//! `subscribe` on that handle, in the contract's own wire format. //! //! ## Compile-Time Safe Keys //! @@ -51,93 +58,20 @@ //! because a self-signed broker cert won't otherwise validate. use aimdb_core::buffer::BufferCfg; -use aimdb_core::{AimDbBuilder, DbResult, Producer, RecordKey, RuntimeContext}; -use aimdb_data_contracts::LinkableRegistrarExt; -use aimdb_mqtt_connector::{MqttLinkExt, MqttOutboundLinkExt}; +use aimdb_core::{AimDbBuilder, DbResult, RecordKey}; +use aimdb_data_contracts::{publish, subscribe}; +use aimdb_mqtt_connector::{MqttConnector, MqttLinkExt, MqttOutboundLinkExt}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use std::sync::Arc; -// Import shared types, monitors, and compile-time safe keys from the common crate +// Shared contracts, keys, thermometers and monitors from the common crate use mqtt_connector_demo_common::{ - command_consumer, temperature_logger, CommandKey, SensorKey, Temperature, TemperatureCommand, + command_consumer, temperature_logger, CommandKey, SensorKey, SimThermometers, Temperature, + TemperatureCommand, Thermometer, }; -// ============================================================================ -// TEMPERATURE PRODUCERS (platform-specific) -// ============================================================================ - -/// Indoor temperature sensor producer -async fn indoor_temp_producer(ctx: RuntimeContext, temperature: Producer) { - let log = ctx.log(); - let time = ctx.time(); - - log.info("🏠 Starting INDOOR temperature producer...\n"); - - for i in 0..5 { - let temp = Temperature::new("indoor-001", 22.0 + (i as f32 * 0.5)); // Indoor temps: 22-24Β°C - - log.info(&format!( - "🏠 Indoor sensor producing: {:.1}Β°C", - temp.celsius - )); - - temperature.produce(temp); - - time.sleep_secs(2).await; - } - - log.info("βœ… Indoor producer finished"); -} - -/// Outdoor temperature sensor producer -async fn outdoor_temp_producer(ctx: RuntimeContext, temperature: Producer) { - let log = ctx.log(); - let time = ctx.time(); - - log.info("🌳 Starting OUTDOOR temperature producer...\n"); - - for i in 0..5 { - let temp = Temperature::new("outdoor-001", 5.0 + (i as f32 * 1.0)); // Outdoor temps: 5-9Β°C (cold!) - - log.info(&format!( - "🌳 Outdoor sensor producing: {:.1}Β°C", - temp.celsius - )); - - temperature.produce(temp); - - time.sleep_secs(2).await; - } - - log.info("βœ… Outdoor producer finished"); -} - -/// Server room temperature sensor producer -async fn server_room_temp_producer(ctx: RuntimeContext, temperature: Producer) { - let log = ctx.log(); - let time = ctx.time(); - - log.info("πŸ–₯️ Starting SERVER ROOM temperature producer...\n"); - - for i in 0..5 { - let temp = Temperature::new("server-room-001", 18.0 + (i as f32 * 0.2)); // Server room: 18-19Β°C (cooled) - - log.info(&format!( - "πŸ–₯️ Server room sensor producing: {:.1}Β°C", - temp.celsius - )); - - temperature.produce(temp); - - time.sleep_secs(2).await; - } - - log.info("βœ… Server room producer finished"); -} - -// ============================================================================ -// MAIN -// ============================================================================ +/// Readings per thermometer before it stops. +const READINGS: usize = 5; #[tokio::main] async fn main() -> DbResult<()> { @@ -154,56 +88,61 @@ async fn main() -> DbResult<()> { println!("Broker: {broker_url}"); println!(); - let mut builder = AimDbBuilder::new().runtime(runtime).with_connector( - aimdb_mqtt_connector::MqttConnector::new(broker_url) - .with_client_id("tokio-demo-multi-sensor"), - ); + let mut builder = AimDbBuilder::new().runtime(runtime); + let mqtt = + builder.connect(MqttConnector::new(broker_url).with_client_id("tokio-demo-multi-sensor")); + let SimThermometers { + indoor, + outdoor, + server_room, + } = SimThermometers::new(); - // Temperature sensors (outbound to MQTT) - using link_address() from key metadata. - // TempIndoor needs per-link QoS/retain, so it stays on the raw builder (the - // escape hatch for per-link options); TempOutdoor/TempServerRoom need no - // extra options, so `.linked_to()` (from `#[derive(Linkable)]`) suffices. + // Temperature sensors (outbound to MQTT), topics from the keys' `#[link_address]`. builder.configure::(SensorKey::TempIndoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .source(indoor_temp_producer) + .input(Thermometer::on(indoor).readings(READINGS)) .tap(temperature_logger) - .link_to(SensorKey::TempIndoor.link_address().unwrap()) - .with_qos(1) - .with_retain(false) - .with_serializer(|_ctx, temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .output( + publish(mqtt, SensorKey::TempIndoor.link_address().unwrap()) + .with_qos(1) + .with_retain(false), + ); }); builder.configure::(SensorKey::TempOutdoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .source(outdoor_temp_producer) - .tap(temperature_logger); - reg.linked_to(SensorKey::TempOutdoor.link_address().unwrap()); + .input(Thermometer::on(outdoor).readings(READINGS)) + .tap(temperature_logger) + .output(publish( + mqtt, + SensorKey::TempOutdoor.link_address().unwrap(), + )); }); builder.configure::(SensorKey::TempServerRoom, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .source(server_room_temp_producer) - .tap(temperature_logger); - reg.linked_to(SensorKey::TempServerRoom.link_address().unwrap()); + .input(Thermometer::on(server_room).readings(READINGS)) + .tap(temperature_logger) + .output(publish( + mqtt, + SensorKey::TempServerRoom.link_address().unwrap(), + )); }); - // Command consumers (inbound from MQTT) - using link_address() from key metadata. - // TempIndoor needs subscribe QoS, so it stays on the raw builder; TempOutdoor - // uses `.linked_from()`. + // Command consumers (inbound from MQTT). builder.configure::(CommandKey::TempIndoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .tap(command_consumer) - .link_from(CommandKey::TempIndoor.link_address().unwrap()) - .with_qos(1) // subscribe QoS via MqttLinkExt - .with_deserializer(|_ctx, data: &[u8]| TemperatureCommand::from_json(data)) - .finish(); + .input(subscribe(mqtt, CommandKey::TempIndoor.link_address().unwrap()).with_qos(1)) + .tap(command_consumer); }); builder.configure::(CommandKey::TempOutdoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .input(subscribe( + mqtt, + CommandKey::TempOutdoor.link_address().unwrap(), + )) .tap(command_consumer); - reg.linked_from(CommandKey::TempOutdoor.link_address().unwrap()); }); println!("Subscribe: mosquitto_sub -h localhost -t 'sensors/#' -v");