Feat: add an example for spmc ring buffer

Cargo.toml

@@ -38,9 +38,10 @@ members = [
     "examples/hello-mailbox",
     "examples/hello-mailbox-async",
     "examples/hello-single-latest-async",
+    "examples/hello-spmc-ring-async",
     "examples/readme-quickstart",
     # Benchmarking infrastructure — host-only, excluded from default-members
-    "aimdb-bench",
+    "aimdb-bench", 
 ]
 exclude = ["_external"]
 resolver = "2"

Makefile

@@ -182,7 +182,7 @@ test:
 
 fmt:
 	@printf "$(GREEN)Formatting code (workspace members only)...$(NC)\n"
-	@for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \
+	@for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async hello-spmc-ring-async aimdb-bench; do \
 		printf "$(YELLOW)  → Formatting $$pkg$(NC)\n"; \
 		cargo fmt -p $$pkg 2>/dev/null || true; \
 	done
@@ -191,7 +191,7 @@ fmt:
 fmt-check:
 	@printf "$(GREEN)Checking code formatting (workspace members only)...$(NC)\n"
 	@FAILED=0; \
-	for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async aimdb-bench; do \
+	for pkg in aimdb-derive aimdb-data-contracts aimdb-core aimdb-client aimdb-embassy-adapter aimdb-tokio-adapter aimdb-wasm-adapter aimdb-sync aimdb-persistence aimdb-persistence-sqlite aimdb-mqtt-connector aimdb-knx-connector aimdb-ws-protocol aimdb-websocket-connector aimdb-uds-connector aimdb-serial-connector aimdb-tcp-connector aimdb-codegen aimdb-cli aimdb-mcp sync-api-demo tokio-mqtt-connector-demo embassy-mqtt-connector-demo tokio-knx-connector-demo embassy-knx-connector-demo embassy-serial-connector-demo embassy-bench-stm32h5 weather-mesh-common weather-hub weather-station-alpha weather-station-beta hello-mailbox hello-mailbox-async hello-single-latest-async hello-spmc-ring-async aimdb-bench; do \
 		printf "$(YELLOW)  → Checking $$pkg$(NC)\n"; \
 		if ! cargo fmt -p $$pkg -- --check 2>&1; then \
 			printf "$(RED)❌ Formatting check failed for $$pkg$(NC)\n"; \
@@ -418,6 +418,8 @@ examples:
 	cargo build --package hello-mailbox-async
 	@printf "$(YELLOW)  → Building hello-single-latest-async$(NC)\n"
 	cargo build --package hello-single-latest-async
+	@printf "$(YELLOW)  → Building hello-spmc-ring-async$(NC)\n"
+	cargo build --package hello-spmc-ring-async
 	@printf "$(YELLOW)  → Building readme-quickstart (compiled README example)$(NC)\n"
 	cargo build --package readme-quickstart
 	@printf "$(GREEN)All examples built successfully!$(NC)\n"

README.md

@@ -131,7 +131,7 @@ docker compose up
 
 | Buffer | Semantics | Use cases |
 | --- | --- | --- |
-| **SPMC Ring** | Bounded stream with independent consumers | Sensor telemetry, event logs |
+| [**SPMC Ring**](examples/hello-spmc-ring-async) | Bounded stream with independent consumers | Sensor telemetry, event logs |
 | [**SingleLatest**](examples/hello-single-latest-async) | Only the current value matters | Feature flags, config, UI state |
 | [**Mailbox**](examples/hello-mailbox) / [**async Mailbox**](examples/hello-mailbox-async)| Latest instruction wins | Device commands, actuation, RPC |
 

examples/hello-spmc-ring-async/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "hello-spmc-ring-async"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+description = "AimDB minimal example demonstrating async SPMC buffer"
+publish = false
+
+[dependencies]
+aimdb-core = { path = "../../aimdb-core", features = ["std"] }
+aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = ["tokio-runtime"] }
+tokio = { workspace = true, features = ["time"] }

examples/hello-spmc-ring-async/README.md

@@ -0,0 +1,41 @@
+# hello-spmc-ring-async: async Spmc Ring buffer demo
+The Spmc Ring buffer supports Single Producer - Multi Consumer pattern. It features a ring buffer with fixed capacity. Each consumer could read the full sequence at its own rate without waiting for consumption states of other consumers.
+
+## When to use
+This Spmc Ring buffer demo is useful in a variety of scenarios where you need to broadcast messages to different types of consumers but you are not sure if one could lag behind others.
+
+## How it works
+The Spmc buffer demo works by simulating:
+- A producer generating a message at fixed interval (struct `Temperature`);
+- A consumer/observer 01 consuming at a rate faster than producer's;
+- A consumer/observer 02 consuming at a rate slower than producer's, and eventually lagging behind (messages dropped), as the ring size is small (10 slots);
+
+## How to run
+From the workspace root, run:
+```
+cargo run -p hello-spmc-ring-async
+```
+**Expected output**
+```
+=== hello-spmc-ring-async: SPMC ring buffer demo ===
+
+Ring size:          10
+Producer at rate    20.0 messages/sec
+Observer 01 at rate 25.0 message/sec
+Observer 02 at rate  6.7 message/sec
+
+
+Observer 01: At 1783932208: -18 celcius
+Observer 02: At 1783932208: -18 celcius
+Observer 01: At 1783932208: -16 celcius
+Observer 01: At 1783932208: -14 celcius
+...
+Observer 02: At 1783932208: -2 celcius
+Observer 01: At 1783932209: 30 celcius
+Observer 01: At 1783932209: 32 celcius
+Observer 01: At 1783932209: 34 celcius
+Observer 02 lagged, dropped 2
+Observer 02: At 1783932209: 4 celcius
+...
+
+```

examples/hello-spmc-ring-async/src/main.rs

@@ -0,0 +1,99 @@
+use std::time::Duration;
+use std::{fmt::Display, sync::Arc};
+
+use aimdb_core::{buffer::BufferCfg, AimDbBuilder, Producer, RuntimeContext};
+use aimdb_core::{Consumer, DbError};
+use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct Temperature {
+    /// Temperature in degrees Celsius
+    pub celcius: f32,
+
+    /// Unix timestamp (milliseconds)
+    pub timestamp: u64,
+}
+
+impl Display for Temperature {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "At {}: {} celcius", self.timestamp, self.celcius)
+    }
+}
+
+// Main function
+#[tokio::main]
+async fn main() -> Result<(), DbError> {
+    println!("=== hello-spmc-ring-async: SPMC ring buffer demo ===\n");
+    println!("Ring size:          10");
+    println!("Producer at rate    20.0 messages/sec");
+    println!("Observer 01 at rate 25.0 messages/sec");
+    println!("Observer 02 at rate  6.7 messages/sec");
+    println!("\n");
+
+    // configuration
+    let adapter = Arc::new(TokioAdapter);
+    let mut builder = AimDbBuilder::new().runtime(adapter);
+
+    // Two observers consuming at different rates
+    builder.configure::<Temperature>("sensor.temp", |reg| {
+        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
+            .source(rollout_source)
+            .tap(rollout_observer01)
+            .tap(rollout_observer02);
+    });
+
+    let (_db, runner) = builder.build().await?;
+    tokio::spawn(runner.run());
+    tokio::time::sleep(Duration::from_millis(6000)).await;
+
+    Ok(())
+}
+
+async fn rollout_source(ctx: RuntimeContext, producer: Producer<Temperature>) {
+    let time = ctx.time();
+    let t0: f32 = -20.0;
+    for i in 0..40 {
+        time.sleep_millis(50).await;
+        // Read wall-clock time through the runtime abstraction rather than
+        // `SystemTime::now()`, so the example stays runtime/sim-friendly.
+        let timestamp = time.unix_time().map(|(secs, _)| secs).unwrap_or(0);
+        publich_rollout(&producer, t0 + (i as f32 + 1.0) * 2.0, timestamp).await;
+    }
+}
+
+async fn publich_rollout(producer: &Producer<Temperature>, t: f32, timestamp: u64) {
+    let temperature = Temperature {
+        celcius: t,
+        timestamp,
+    };
+    producer.produce(temperature);
+}
+
+async fn rollout_observer01(ctx: RuntimeContext, consumer: Consumer<Temperature>) {
+    let mut reader = consumer.subscribe();
+    let time = ctx.time();
+
+    while let Ok(t) = reader.recv().await {
+        println!("Observer 01: {}", t);
+        time.sleep_millis(40).await;
+    }
+}
+
+async fn rollout_observer02(ctx: RuntimeContext, consumer: Consumer<Temperature>) {
+    let mut reader = consumer.subscribe();
+    let time = ctx.time();
+
+    // Consume at slower rate, must miss data
+    loop {
+        match reader.recv().await {
+            Ok(t) => {
+                println!("Observer 02: {}", t);
+                time.sleep_millis(150).await;
+            }
+            Err(DbError::BufferLagged { lag_count, .. }) => {
+                println!("Observer 02 lagged, dropped {lag_count}");
+            }
+            Err(_) => break,
+        };
+    }
+}