From 23e505f348806f47f9987050566de95f67cb3c94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Mon, 5 Oct 2026 19:13:08 +0000 Subject: [PATCH] =?UTF-8?q?feat(knx)!:=20connection=20task=20dispatches=20?= =?UTF-8?q?and=20pulls;=20drop=20Channels=20(design=20054=20=C2=A74.9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The KNX connection task now dispatches inbound telegrams through an `InboundDispatch` and pulls outbound values from `OutboundRoutes` while the tunnel is connected. `build()` returns that one future; core runs no pump for KNX any more. Breaking: `KnxConnector::new(binder, delay, gateway_url)` loses the `&'static Channels` argument and the `N` parameter. `Channels`, `TelegramChannel`, `CommandChannel`, `DEFAULT_QUEUE`, `KnxSink` and `KnxSource` are gone. Values produced during connect or backoff wait in their record buffers and go out after the handshake. An invalid group address or oversize payload is logged and counted as rejected. - client: `TelegramSink::deliver(&str, &[u8])` over `InboundDispatch`, group address formatted on the stack; `CommandSource` and the embassy-sync channel glue removed. - Tests: one future with or without outbound routes; the client tests build `OutboundRoutes` from a test db; an end-to-end round trip through records against a fake gateway; a value produced during connect is sent after the handshake. `topic_provider_tests.rs` becomes `topic_writer_tests.rs`; `shared_channel_on_std.rs` and the `critical-section` dev-dependency are removed. - Demos, `aimdb-codegen`'s KNX templates, the usage guide and the KNX README drop the channels. Co-Authored-By: Claude Opus 5.5 --- aimdb-codegen/src/rust.rs | 35 +- aimdb-knx-connector/Cargo.toml | 18 +- aimdb-knx-connector/README.md | 19 +- aimdb-knx-connector/src/client.rs | 337 +++++----- aimdb-knx-connector/src/connector.rs | 272 ++------ aimdb-knx-connector/src/lib.rs | 14 +- aimdb-knx-connector/src/tunnel.rs | 12 +- .../tests/shared_channel_on_std.rs | 53 -- .../tests/topic_provider_tests.rs | 590 ------------------ .../tests/topic_writer_tests.rs | 405 ++++++++++++ docs/aimdb-usage-guide.md | 6 +- .../embassy-knx-connector-demo/src/main.rs | 12 +- examples/tokio-knx-connector-demo/README.md | 4 - examples/tokio-knx-connector-demo/src/main.rs | 9 +- 14 files changed, 697 insertions(+), 1089 deletions(-) delete mode 100644 aimdb-knx-connector/tests/shared_channel_on_std.rs delete mode 100644 aimdb-knx-connector/tests/topic_provider_tests.rs create mode 100644 aimdb-knx-connector/tests/topic_writer_tests.rs diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs index bd2c42b7..6b9940ec 100644 --- a/aimdb-codegen/src/rust.rs +++ b/aimdb-codegen/src/rust.rs @@ -251,7 +251,7 @@ pub fn generate_main_rs(state: &ArchitectureState, binary_name: &str) -> Option< .filter_map(|c| match c.protocol.as_str() { "mqtt" => Some(quote! { use aimdb_mqtt_connector::MqttConnector; }), "knx" => Some(quote! { - use aimdb_knx_connector::{Channels, KnxConnector}; + use aimdb_knx_connector::KnxConnector; use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; }), "ws" => Some(quote! { use aimdb_websocket_connector::WebSocketConnector; }), @@ -269,18 +269,13 @@ pub fn generate_main_rs(state: &ArchitectureState, binary_name: &str) -> Option< let default = &c.default; let ctor: TokenStream = match c.protocol.as_str() { "mqtt" => quote! { MqttConnector::new(&#var_ident) }, - // The adapter owns the socket and the clock; the channels are - // the binary's, in a block-scoped `static`. + // The adapter owns the socket and the clock. "knx" => quote! { - { - static KNX_CHANNELS: Channels = Channels::new(); - KnxConnector::new( - TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED), - TokioDelay, - &#var_ident, - &KNX_CHANNELS, - ) - } + KnxConnector::new( + TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED), + TokioDelay, + &#var_ident, + ) }, "ws" => quote! { WebSocketConnector::new() @@ -1411,7 +1406,7 @@ pub fn generate_hub_main_rs(state: &ArchitectureState) -> String { } if has_knx { v.push(quote! { - use aimdb_knx_connector::{Channels, KnxConnector}; + use aimdb_knx_connector::KnxConnector; use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; }); } @@ -1456,15 +1451,11 @@ pub fn generate_hub_main_rs(state: &ArchitectureState) -> String { } if has_knx { v.push(quote! { - .with_connector({ - static KNX_CHANNELS: Channels = Channels::new(); - KnxConnector::new( - TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED), - TokioDelay, - &knx_gateway, - &KNX_CHANNELS, - ) - }) + .with_connector(KnxConnector::new( + TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED), + TokioDelay, + &knx_gateway, + )) }); } if has_ws { diff --git a/aimdb-knx-connector/Cargo.toml b/aimdb-knx-connector/Cargo.toml index 34863fd4..d50e4329 100644 --- a/aimdb-knx-connector/Cargo.toml +++ b/aimdb-knx-connector/Cargo.toml @@ -21,12 +21,12 @@ default = ["aimdb-core/alloc"] # feature here — which is what lets a third runtime (FreeRTOS/lwIP) use this # crate with no edit to it. # -# `embassy-sync` is not an Embassy dependency in the executor sense: it is -# no_std, no_alloc, and pulls no executor, and its `Channel` is this connector's -# public queue type (`Channels`) on every runtime. +# `embassy-sync` no longer backs a public type here (the connection task pulls +# from the record buffers directly); it is dropped together with the old +# connector SPI. connector = [ "aimdb-core/alloc", - "aimdb-core/connector-session", # `pump_sink`/`pump_source`/`Source`/`Payload` + "aimdb-core/connector-session", # `DatagramBinder`/`Datagram`/`Delay` "embassy-sync", ] @@ -49,10 +49,9 @@ embassy-runtime = ["connector"] # the **final binary** may choose one: a library enabling it would hand every # downstream binary a duplicate-symbol link error with no way to opt out. # -# So this stays off by default and out of `connector`/`std`. This crate's own -# tests get the impl through a dev-dependency; a std binary that instantiates a -# `CriticalSectionRawMutex` channel and has no other impl in its graph can -# either depend on `critical-section` with `features = ["std"]` directly (the +# So this stays off by default and out of `connector`/`std`. A std binary that +# instantiates a `CriticalSectionRawMutex` channel and has no other impl in its +# graph can either depend on `critical-section` with `features = ["std"]` directly (the # documented way) or enable this feature. On Embassy the HAL (cortex-m / # embassy-rp / ...) already provides one. critical-section-std-impl = ["critical-section/std"] @@ -92,9 +91,6 @@ heapless = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["full"] } -# Test binaries are binaries: selecting the impl here lets `CriticalSectionRawMutex` -# link for this crate's tests without imposing that choice on consumers. -critical-section = { version = "1.2", features = ["std"] } tokio-test = "0.4" aimdb-tokio-adapter = { path = "../aimdb-tokio-adapter", features = [ "tokio-runtime", diff --git a/aimdb-knx-connector/README.md b/aimdb-knx-connector/README.md index 78add83b..9587c430 100644 --- a/aimdb-knx-connector/README.md +++ b/aimdb-knx-connector/README.md @@ -45,7 +45,7 @@ We're working with upstream to get these changes merged. Once published, the pat ## Quick Start (Tokio) ```rust -use aimdb_knx_connector::{Channels, KnxConnector}; +use aimdb_knx_connector::KnxConnector; use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; use aimdb_tokio_adapter::TokioAdapter; use std::net::Ipv4Addr; @@ -55,11 +55,6 @@ struct LightState { is_on: bool, } -// The connector's queues. `'static` because the connection task and the pumps -// are spawned as `'static` futures; a `StaticCell` supplies this on an MCU. -// One pair per connector — do not share it between two `KnxConnector`s. -static CHANNELS: Channels = Channels::new(); - #[tokio::main] async fn main() -> Result<(), Box> { let db = AimDbBuilder::new() @@ -70,7 +65,6 @@ async fn main() -> Result<(), Box> { TokioNet::udp(Ipv4Addr::UNSPECIFIED), TokioDelay, "knx://192.168.1.19:3671", - &CHANNELS, )) .configure::(|reg| { reg.buffer(BufferCfg::SingleLatest) @@ -96,14 +90,11 @@ async fn main() -> Result<(), Box> { ## Quick Start (Embassy) -The same constructor — only the binder and clock change, and the channels come -from a `static` instead of being sized at runtime: +The same constructor — only the binder and clock change: ```rust use aimdb_embassy_adapter::net::{EmbassyDelay, EmbassyNet}; -use aimdb_knx_connector::{Channels, KnxConnector}; - -static CHANNELS: Channels<32> = Channels::new(); +use aimdb_knx_connector::KnxConnector; let binder = EmbassyNet::udp(stack, rx_meta, rx_buf, tx_meta, tx_buf); @@ -113,12 +104,10 @@ let builder = AimDbBuilder::new() binder, EmbassyDelay, "knx://192.168.1.19:3671", - &CHANNELS, )); ``` -Enable `features = ["connector"], default-features = false` — no -`critical-section-std-impl`, since the HAL already provides an impl. See +Enable `features = ["connector"], default-features = false`. See `examples/embassy-knx-connector-demo/` for the full wiring. ## Group Address Format diff --git a/aimdb-knx-connector/src/client.rs b/aimdb-knx-connector/src/client.rs index 6b01384e..f13d33bf 100644 --- a/aimdb-knx-connector/src/client.rs +++ b/aimdb-knx-connector/src/client.rs @@ -4,21 +4,22 @@ //! The clock stays `RuntimeOps::now_nanos`, a plain call; only *sleeping* goes //! through [`Delay`], so nothing is boxed per loop iteration. //! -//! The `embassy-sync` and `embassy-futures` types below are executor-independent -//! — neither pulls an executor, and both build on std — so they back this task -//! on either runtime. +//! The task dispatches inbound telegrams into their records through an +//! [`InboundDispatch`](aimdb_core::InboundDispatch) and pulls outbound values +//! from an [`OutboundRoutes`] while the tunnel is connected. `embassy-futures`' +//! select is executor-independent, so the task runs on either runtime. -use alloc::string::{String, ToString}; use alloc::sync::Arc; use alloc::vec::Vec; -use core::future::Future; +use core::fmt::Write as _; +use core::future::poll_fn; use core::net::SocketAddr; +use core::task::Poll; use core::time::Duration; -use aimdb_core::session::{ - Datagram, DatagramBinder, Delay, Payload, TransportError, TransportResult, -}; -use aimdb_core::{log_debug, log_error, log_warn, RuntimeOps}; +use aimdb_core::connector::TopicBuf; +use aimdb_core::session::{Datagram, DatagramBinder, Delay, TransportError, TransportResult}; +use aimdb_core::{log_debug, log_error, log_warn, OutboundRoutes, RouteId, RuntimeOps}; use crate::tunnel::{ drain_actions, GroupWrite, LocalEndpoint, Millis, TunnelConfig, TunnelEngine, TunnelIo, @@ -30,22 +31,24 @@ const BIND_RETRY: Duration = Duration::from_secs(5); /// Per-datagram receive buffer; a KNXnet/IP frame fits comfortably. const RECV_BUF: usize = 512; -/// Where parsed telegrams go — an `embassy_sync` channel on either runtime. +/// Where parsed telegrams go: an [`InboundDispatch`](aimdb_core::InboundDispatch) +/// in the connector. /// -/// Non-blocking by contract: a full sink drops rather than stalling the -/// protocol loop. +/// Non-blocking by contract: delivery never stalls the protocol loop. pub trait TelegramSink { - /// Enqueue one `(group-address, payload)`. `false` if it was dropped. - fn try_send(&self, topic: String, payload: Payload) -> bool; + /// Deliver one telegram for group address `topic`. + fn deliver(&self, topic: &str, payload: &[u8]); } -/// Where outbound commands come from, the dual of [`TelegramSink`]. -pub trait CommandSource { - /// Yield the next command. Parks forever once no producer remains, so the - /// select arm goes quiet instead of ending the task. - fn recv(&mut self) -> impl Future + Send + '_; +impl TelegramSink for aimdb_core::InboundDispatch { + fn deliver(&self, topic: &str, payload: &[u8]) { + self.dispatch(topic, payload); + } } +/// The longest group address, `31/7/255`, is 8 bytes. +const GROUP_ADDRESS_LEN: usize = 16; + /// The socket-side glue for [`drain_actions`], written once against /// [`Datagram`] instead of once per runtime. struct NeutralIo<'a, U, S> { @@ -77,8 +80,10 @@ where fn forward(&mut self, addr: GroupAddress, payload: Vec) { log_debug!("KNX telegram: {} ({} bytes)", addr, payload.len()); - if !self.sink.try_send(addr.to_string(), Payload::from(payload)) { - log_warn!("KNX inbound: dropping telegram for {} (sink full)", addr); + let mut storage = [0u8; GROUP_ADDRESS_LEN]; + let mut topic = TopicBuf::new(&mut storage); + if write!(topic, "{addr}").is_ok() { + self.sink.deliver(topic.as_str(), &payload); } } @@ -87,20 +92,44 @@ where } } +/// Hand the value `outbound` staged for route `id` to the engine as a group +/// write. An invalid group address or an oversize payload is logged, counted +/// as rejected in the route's `RouteStats`, and skipped. +fn send_staged(engine: &mut TunnelEngine, outbound: &mut OutboundRoutes, id: RouteId, now: Millis) { + let command = { + let Some(msg) = outbound.take_staged() else { + return; + }; + GroupWrite::try_new(msg.topic, msg.payload.as_slice()) + }; + match command { + Ok(command) => { + let _ = engine.handle_command(command, now); + } + Err(_e) => { + log_warn!( + "KNX outbound: skipping a value for route '{}': {:?}", + outbound.routes()[id].default_topic, + _e + ); + outbound.reject(id); + } + } +} + /// Drive the engine over one socket's lifetime; returns when it asks for a reset. -async fn drive_connection( +async fn drive_connection( engine: &mut TunnelEngine, socket: &mut U, gateway: SocketAddr, runtime: &Arc, delay: &D, sink: &S, - commands: &mut C, + outbound: &mut OutboundRoutes, ) where U: Datagram + Send, D: Delay, S: TelegramSink + Sync, - C: CommandSource, { // Executor-independent despite the name: `embassy-futures` has no // dependencies and its select is pure `core::task`. @@ -129,9 +158,12 @@ async fn drive_connection( // one — unlike the `tokio::select!` this task replaces, which picked among // the ready arms at random. With a fixed order, sustained inbound traffic // means the first arm is ready on every pass and the command arm is never - // reached, so outbound `GroupWrite`s stall until the channel drops them. + // reached, so outbound values stall in their record buffers. // Swapping the two contended arms each pass restores that fairness. let mut inbound_first = true; + // Latched on `Ready(None)` (every route closed, or none): passing it + // through would resolve the arm on every pass and the loop would spin. + let mut outbound_done = false; loop { engine.poll(now_ms()); @@ -151,34 +183,37 @@ async fn drive_connection( let deadline = delay.sleep(Duration::from_millis(sleep_ms)); let mut recv_buf = [0u8; RECV_BUF]; - // Only drain commands while connected: during connect and backoff the - // arm stays pending, so commands queue and flush once the handshake - // completes — as both hand-written clients do. + // Only pull values while connected: during connect and backoff the + // arm stays pending, so values wait in their record buffers and go out + // once the handshake completes. A value leaves its buffer only when + // the arm resolves, so a losing arm takes nothing. let connected = engine.is_connected(); - let cmd_arm = async { - if connected { - commands.recv().await - } else { - core::future::pending().await + let cmd_arm = poll_fn(|cx| { + if !connected || outbound_done { + return Poll::Pending; } - }; + match outbound.poll_stage(cx) { + Poll::Ready(Some(id)) => Poll::Ready(id), + Poll::Ready(None) => { + outbound_done = true; + Poll::Pending + } + Poll::Pending => Poll::Pending, + } + }); // The deadline arm stays last in both orders: it only ever asks for a // `poll` the loop top would reach anyway. if inbound_first { match select3(socket.recv_from(&mut recv_buf), cmd_arm, deadline).await { Either3::First(r) => apply_inbound(engine, &recv_buf, r, now_ms()), - Either3::Second(cmd) => { - let _ = engine.handle_command(cmd, now_ms()); - } + Either3::Second(id) => send_staged(engine, outbound, id, now_ms()), // Woken for the engine deadline; `poll` at the loop top fires it. Either3::Third(()) => {} } } else { match select3(cmd_arm, socket.recv_from(&mut recv_buf), deadline).await { - Either3::First(cmd) => { - let _ = engine.handle_command(cmd, now_ms()); - } + Either3::First(id) => send_staged(engine, outbound, id, now_ms()), Either3::Second(r) => apply_inbound(engine, &recv_buf, r, now_ms()), Either3::Third(()) => {} } @@ -192,18 +227,17 @@ async fn drive_connection( /// Binds a socket, advertises its real local endpoint when the stack exposes /// one, drives the shared [`TunnelEngine`] over that socket's lifetime, then /// rebinds after the engine's backoff. -pub async fn connection_task( +pub async fn connection_task( binder: B, gateway: SocketAddr, runtime: Arc, delay: D, sink: S, - mut commands: C, + mut outbound: OutboundRoutes, ) where B: DatagramBinder, D: Delay, S: TelegramSink + Sync, - C: CommandSource, { let now_ms = || runtime.now_nanos() / 1_000_000; let mut engine = TunnelEngine::new(TunnelConfig::default(), now_ms()); @@ -267,7 +301,7 @@ pub async fn connection_task( &runtime, &delay, &sink, - &mut commands, + &mut outbound, ) .await; @@ -280,45 +314,15 @@ pub async fn connection_task( } } -/// Channel bridges over `embassy_sync`, which is executor-independent, so the -/// same types back the task on both runtimes. -#[cfg(feature = "connector")] -pub mod shared_channel { - use super::{CommandSource, GroupWrite, Payload, TelegramSink}; - use alloc::string::String; - use core::future::Future; - use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; - use embassy_sync::channel::{Receiver, Sender}; - - /// Sending half of the inbound-telegram channel. - pub struct ChannelSink<'a, const N: usize>( - pub Sender<'a, CriticalSectionRawMutex, (String, Payload), N>, - ); - - impl TelegramSink for ChannelSink<'_, N> { - fn try_send(&self, topic: String, payload: Payload) -> bool { - self.0.try_send((topic, payload)).is_ok() - } - } - - /// Receiving half of the outbound-command channel. - pub struct ChannelCommands<'a, const N: usize>( - pub Receiver<'a, CriticalSectionRawMutex, GroupWrite, N>, - ); - - impl CommandSource for ChannelCommands<'_, N> { - fn recv(&mut self) -> impl Future + Send + '_ { - self.0.receive() - } - } -} - #[cfg(all(test, feature = "std"))] mod tests { use super::*; + use aimdb_core::buffer::BufferCfg; use aimdb_core::session::TransportError; + use aimdb_core::AimDbBuilder; use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; - use aimdb_tokio_adapter::TokioAdapter; + use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + use core::future::Future; use core::pin::Pin; use std::net::Ipv4Addr; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -326,22 +330,25 @@ mod tests { /// Collects forwarded telegrams; `Sync`, as [`TelegramSink`] requires. #[derive(Default)] - struct VecSink(Mutex>); + struct VecSink(Mutex)>>); impl TelegramSink for VecSink { - fn try_send(&self, topic: String, payload: Payload) -> bool { - self.0.lock().expect("sink mutex").push((topic, payload)); - true + fn deliver(&self, topic: &str, payload: &[u8]) { + self.0 + .lock() + .expect("sink mutex") + .push((topic.into(), payload.into())); } } - /// No outbound producer: the command arm never fires. - struct NoCommands; - - impl CommandSource for NoCommands { - fn recv(&mut self) -> impl Future + Send + '_ { - core::future::pending() - } + /// No outbound routes: the command arm never fires. + async fn no_outbound() -> OutboundRoutes { + let (db, _runner) = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .build() + .await + .expect("build db"); + OutboundRoutes::new(&db, "knx").expect("outbound routes") } fn runtime() -> Arc { @@ -354,15 +361,15 @@ mod tests { /// `Pin + Send + 'static>>`. Everything that /// declares `+ Send` on a return type does so to make this line compile for /// a *generic* task. - #[test] - fn unified_task_is_boxable_as_the_runners_send_future() { + #[tokio::test] + async fn unified_task_is_boxable_as_the_runners_send_future() { let task = connection_task( TokioNet::udp(Ipv4Addr::LOCALHOST), "127.0.0.1:3671".parse().expect("gateway addr"), runtime(), TokioDelay, VecSink::default(), - NoCommands, + no_outbound().await, ); let _boxed: Pin + Send + 'static>> = Box::pin(task); } @@ -384,7 +391,7 @@ mod tests { runtime(), TokioDelay, VecSink::default(), - NoCommands, + no_outbound().await, )); let mut buf = [0u8; 128]; @@ -430,7 +437,7 @@ mod tests { runtime(), TokioDelay, VecSink::default(), - NoCommands, + no_outbound().await, )); let mut buf = [0u8; 128]; @@ -451,39 +458,10 @@ mod tests { task.abort(); } - /// The unified task on Tokio, moving real telegrams through the *same* - /// `embassy_sync` channel types the MCU uses: a full handshake, an inbound - /// telegram with its ACK, and an outbound command. - #[tokio::test] - async fn shared_embassy_channels_carry_telegrams_on_tokio() { - use super::shared_channel::{ChannelCommands, ChannelSink}; - use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; - use embassy_sync::channel::Channel; - - const N: usize = 8; - // Leaked for `'static` borrows, as a `StaticCell` gives on the MCU. - let inbound: &'static Channel = - Box::leak(Box::new(Channel::new())); - let commands: &'static Channel = - Box::leak(Box::new(Channel::new())); - - let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0") - .await - .expect("bind gateway"); - let gateway_addr = gateway.local_addr().expect("gateway addr"); - - let task = tokio::spawn(connection_task( - TokioNet::udp(Ipv4Addr::LOCALHOST), - gateway_addr, - runtime(), - TokioDelay, - ChannelSink::(inbound.sender()), - ChannelCommands::(commands.receiver()), - )); - - let mut buf = [0u8; 1024]; - - // Handshake. + /// Answer the client's CONNECT_REQUEST with channel 7; returns the + /// client's address. + async fn accept_connect(gateway: &tokio::net::UdpSocket) -> SocketAddr { + let mut buf = [0u8; 128]; let (_, client_addr) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) .await .expect("no CONNECT_REQUEST") @@ -498,8 +476,63 @@ mod tests { .send_to(&connect_response, client_addr) .await .expect("send CONNECT_RESPONSE"); + client_addr + } + + /// Wait for a TUNNELING_REQUEST and check it writes 1 to group 1/0/8. + async fn expect_write_to_1_0_8(gateway: &tokio::net::UdpSocket) { + let mut buf = [0u8; 128]; + let (len, _) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) + .await + .expect("no TUNNELING_REQUEST") + .expect("recv_from"); + assert_eq!(u16::from_be_bytes([buf[2], buf[3]]), 0x0420); + assert_eq!(&buf[16..18], &[0x08, 0x08], "cEMI destination = 1/0/8"); + assert_eq!(buf[len - 1], 0x81, "APCI GroupValueWrite | value 1"); + } + + /// A db whose `knx` connector tunnels to `gateway`: record `in` reads group + /// 1/0/7 and record `out` writes group 1/0/8. + async fn knx_db(gateway: SocketAddr) -> (aimdb_core::AimDb, aimdb_core::builder::AimDbRunner) { + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(crate::KnxConnector::new( + TokioNet::udp(Ipv4Addr::LOCALHOST), + TokioDelay, + format!("knx://{gateway}"), + )); + builder.configure::("in", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_from("knx://1/0/7") + .with_deserializer(|_ctx, data: &[u8]| { + data.first().copied().ok_or_else(|| String::from("empty")) + }) + .finish(); + }); + builder.configure::("out", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_to("knx://1/0/8") + .with_serializer(|_ctx, v: &u8| Ok(vec![*v])) + .finish(); + }); + builder.build().await.expect("build db") + } + + /// The connector end to end: a full handshake, an inbound telegram + /// dispatched into its record with its ACK, and a produced value on the + /// wire as a group write. + #[tokio::test] + async fn telegrams_round_trip_through_records() { + let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("bind gateway"); + let (db, runner) = knx_db(gateway.local_addr().expect("gateway addr")).await; + let mut inbound = db.subscribe::("in").expect("subscribe"); + let task = tokio::spawn(runner.run()); - // Inbound telegram -> ACK on the wire, payload on the shared channel. + let client_addr = accept_connect(&gateway).await; + + // Inbound telegram -> ACK on the wire, value in the record. let cemi = [ 0x29, 0x00, 0xBC, 0xE0, 0x00, 0x00, 0x08, 0x07, 0x01, 0x00, 0x81, ]; @@ -513,6 +546,7 @@ mod tests { .await .expect("send telegram"); + let mut buf = [0u8; 128]; tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) .await .expect("no TUNNELING_ACK") @@ -520,29 +554,32 @@ mod tests { assert_eq!(u16::from_be_bytes([buf[2], buf[3]]), 0x0421); assert_eq!(buf[8], 42, "sequence echoed"); - let (topic, payload) = tokio::time::timeout(RECV_TIMEOUT, inbound.receive()) + let value = tokio::time::timeout(RECV_TIMEOUT, inbound.recv()) .await - .expect("no telegram reached the embassy-sync channel"); - assert_eq!(topic, "1/0/7"); - assert_eq!(&payload[..], &[0x01]); - - // Outbound: a command through the shared channel reaches the wire. - let mut data = heapless::Vec::new(); - data.push(0x01).expect("push"); - commands - .send(GroupWrite { - group_addr: "1/0/8".parse().expect("group address"), - data, - }) - .await; + .expect("no telegram reached the record") + .expect("recv"); + assert_eq!(value, 0x01); - let (len, _) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) + // Outbound: a produced value reaches the wire. + db.producer::("out").expect("producer").produce(1); + expect_write_to_1_0_8(&gateway).await; + + task.abort(); + } + + /// A value produced while the tunnel is still connecting waits in its + /// record buffer and goes out once the handshake completes. + #[tokio::test] + async fn a_value_produced_during_connect_is_sent_after_the_handshake() { + let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0") .await - .expect("no TUNNELING_REQUEST") - .expect("recv_from"); - assert_eq!(u16::from_be_bytes([buf[2], buf[3]]), 0x0420); - assert_eq!(&buf[16..18], &[0x08, 0x08], "cEMI destination = 1/0/8"); - assert_eq!(buf[len - 1], 0x81, "APCI GroupValueWrite | value 1"); + .expect("bind gateway"); + let (db, runner) = knx_db(gateway.local_addr().expect("gateway addr")).await; + db.producer::("out").expect("producer").produce(1); + let task = tokio::spawn(runner.run()); + + accept_connect(&gateway).await; + expect_write_to_1_0_8(&gateway).await; task.abort(); } @@ -652,7 +689,7 @@ mod tests { runtime(), TokioDelay, VecSink::default(), - NoCommands, + no_outbound().await, )); // Must exceed the two `BIND_RETRY` sleeps the task waits out. `RECV_TIMEOUT` @@ -704,7 +741,7 @@ mod tests { runtime(), TokioDelay, VecSink::default(), - NoCommands, + no_outbound().await, )); // Cycle 1: the socket knows its address, so the HPAI is explicit. diff --git a/aimdb-knx-connector/src/connector.rs b/aimdb-knx-connector/src/connector.rs index f11bf947..9e9edfd5 100644 --- a/aimdb-knx-connector/src/connector.rs +++ b/aimdb-knx-connector/src/connector.rs @@ -2,9 +2,9 @@ //! //! Generic over core's [`DatagramBinder`](aimdb_core::session::DatagramBinder) //! and [`Delay`](aimdb_core::session::Delay), so the adapter owns the UDP -//! socket and the clock while this crate owns the tunnelling protocol. -//! The channels between the pumps and the connection task are `embassy_sync`, -//! which is executor-independent, so one wiring serves both runtimes. +//! socket and the clock while this crate owns the tunnelling protocol. The one +//! connection task dispatches inbound telegrams into their records and pulls +//! outbound values from their record buffers itself; core runs no pump for it. use alloc::boxed::Box; use alloc::string::String; @@ -16,119 +16,31 @@ use core::net::SocketAddr; use core::pin::Pin; use aimdb_core::connector::{ConnectorBuilder, ConnectorUrl}; -use aimdb_core::session::{pump_sink, pump_source, Payload}; -use aimdb_core::transport::{Connector, ConnectorConfig, PublishError}; -use aimdb_core::{log_info, AimDb, DbError, DbResult, RuntimeOps}; +use aimdb_core::{log_info, AimDb, DbError, DbResult, InboundDispatch, OutboundRoutes, RuntimeOps}; -use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; -use embassy_sync::channel::Channel; - -use crate::client::{ - connection_task, - shared_channel::{ChannelCommands, ChannelSink}, -}; -use crate::tunnel::GroupWrite; +use crate::client::connection_task; type BoxFuture = Pin + Send + 'static>>; /// Default KNXnet/IP tunnelling port. const DEFAULT_PORT: u16 = 3671; -/// Capacity of the command and telegram channels. -/// -/// A const generic rather than a builder setter: the MCU allocates these in a -/// `static`, where the size must be a constant. -pub const DEFAULT_QUEUE: usize = 32; - -/// The inbound-telegram channel type. -pub type TelegramChannel = Channel; -/// The outbound-command channel type. -pub type CommandChannel = Channel; - -/// Outbound half: `pump_sink` hands each serialized record here. -struct KnxSink<'a, const N: usize> { - commands: &'a CommandChannel, -} - -impl Connector for KnxSink<'_, N> { - fn publish( - &self, - destination: &str, - _config: &ConnectorConfig, - payload: &[u8], - ) -> Pin> + Send + '_>> { - // Validation shared with the connection task (same checks, same order). - let command = GroupWrite::try_new(destination, payload); - Box::pin(async move { - self.commands.send(command?).await; - Ok(()) - }) - } -} - -/// Inbound source drained by `pump_source`. -struct KnxSource<'a, const N: usize> { - telegrams: &'a TelegramChannel, -} - -impl aimdb_core::session::Source for KnxSource<'_, N> { - fn next(&mut self) -> aimdb_core::session::BoxFut<'_, Option<(String, Payload)>> { - Box::pin(async move { Some(self.telegrams.receive().await) }) - } -} - /// KNX/IP tunnelling connector over an adapter's datagram transport. -/// -/// `N` sizes both the command and telegram channels. -pub struct KnxConnector { +pub struct KnxConnector { binder: B, delay: D, gateway_url: String, - channels: &'static Channels, -} - -/// The channel pair, held for the process lifetime. -/// -/// `'static` because the connection task and the pumps are spawned as -/// `'static` futures; a `StaticCell` supplies this on the MCU and a `static` -/// item on a host, matching design 037's allocate-at-build model. -pub struct Channels { - telegrams: TelegramChannel, - commands: CommandChannel, } -impl Default for Channels { - fn default() -> Self { - Self::new() - } -} - -impl Channels { - /// A fresh, empty channel pair. - pub const fn new() -> Self { - Self { - telegrams: Channel::new(), - commands: Channel::new(), - } - } -} - -impl KnxConnector { +impl KnxConnector { /// Connect to the KNX/IP gateway at `gateway_url` (`knx://host:port`). /// - /// `binder` and `delay` come from an adapter; `channels` is the caller's - /// `'static` channel pair. - pub fn new( - binder: B, - delay: D, - gateway_url: impl Into, - channels: &'static Channels, - ) -> Self { + /// `binder` and `delay` come from an adapter. + pub fn new(binder: B, delay: D, gateway_url: impl Into) -> Self { Self { binder, delay, gateway_url: gateway_url.into(), - channels, } } @@ -153,7 +65,7 @@ impl KnxConnector { } } -impl ConnectorBuilder for KnxConnector +impl ConnectorBuilder for KnxConnector where B: aimdb_core::session::DatagramBinder + Clone + Send + Sync + 'static, D: aimdb_core::session::Delay + Clone + Send + Sync + 'static, @@ -167,32 +79,20 @@ where log_info!("Creating KNX connector for gateway {}", gateway); let runtime: Arc = db.runtime_ops(); - let channels = self.channels; + // Built here, so routes are subscribed before the task first runs: + // a command produced while the tunnel connects waits in its record + // buffer and goes out once the handshake completes. + let inbound = InboundDispatch::new(db, "knx", &aimdb_core::ExactGrammar)?; + let outbound = OutboundRoutes::new(db, "knx")?; let task: BoxFuture = Box::pin(connection_task( self.binder.clone(), gateway, runtime, self.delay.clone(), - ChannelSink::(channels.telegrams.sender()), - ChannelCommands::(channels.commands.receiver()), + inbound, + outbound, )); - - let mut futures: Vec = vec![task]; - futures.extend(pump_source( - db, - db.inbound_router("knx", &aimdb_core::ExactGrammar)?, - KnxSource:: { - telegrams: &channels.telegrams, - }, - )); - futures.extend(pump_sink( - db, - "knx", - Arc::new(KnxSink:: { - commands: &channels.commands, - }), - )); - Ok(futures) + Ok(vec![task]) }) } @@ -222,8 +122,6 @@ mod tests { use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use std::net::Ipv4Addr; - static CHANNELS: Channels<8> = Channels::new(); - async fn db() -> AimDb { let mut builder = AimDbBuilder::new().runtime(Arc::new(TokioAdapter)); builder.configure::("light", |reg| { @@ -236,11 +134,10 @@ mod tests { #[tokio::test] async fn an_unparsable_gateway_fails_the_build() { let db = db().await; - let connector = KnxConnector::<_, _, 8>::new( + let connector = KnxConnector::new( TokioNet::udp(Ipv4Addr::LOCALHOST), TokioDelay, "knx://not-an-ip:3671", - &CHANNELS, ); let Err(err) = connector.build(&db).await else { panic!("a hostname must be rejected: it is never resolved"); @@ -251,22 +148,18 @@ mod tests { ); } - /// A db with one inbound and one outbound `knx` route, so both pumps have - /// something to contribute. + /// A db with one inbound and one outbound `knx` route. /// /// A connector must be registered or the builder rejects the routes ("no - /// connector registered for scheme 'knx'"). It gets its own channel pair: - /// two connectors sharing one would split the command queue between them. - /// Nothing here is ever driven — `build` only collects futures. + /// connector registered for scheme 'knx'"). Nothing here is ever driven — + /// `build` only collects futures. async fn routed_db() -> AimDb { - static REGISTERED: Channels<8> = Channels::new(); let mut builder = AimDbBuilder::new() .runtime(Arc::new(TokioAdapter)) - .with_connector(KnxConnector::<_, _, 8>::new( + .with_connector(KnxConnector::new( TokioNet::udp(Ipv4Addr::LOCALHOST), TokioDelay, "knx://127.0.0.1:3671", - ®ISTERED, )); builder.configure::("switch", |reg| { reg.buffer(BufferCfg::SingleLatest) @@ -286,90 +179,51 @@ mod tests { } /// The connector registers under the `knx` scheme and contributes exactly - /// the connection task, one `pump_source`, and one publisher per outbound - /// route. - /// - /// The count is asserted, not just non-emptiness: `pump_source` yields one - /// future unconditionally and `pump_sink` one per outbound route, so a pump - /// silently dropping out of `build` is the failure this catches, and - /// `!is_empty()` would not — the connection task alone satisfies that. + /// one future, the connection task, which dispatches inbound telegrams and + /// pulls outbound values itself. #[tokio::test] - async fn build_yields_the_connection_task_and_pumps() { - static CH: Channels<8> = Channels::new(); - let db = routed_db().await; - let connector = KnxConnector::<_, _, 8>::new( - TokioNet::udp(Ipv4Addr::LOCALHOST), - TokioDelay, - "knx://127.0.0.1:3671", - &CH, - ); - assert_eq!(ConnectorBuilder::scheme(&connector), "knx"); - - let futures = connector.build(&db).await.expect("build"); - assert_eq!( - futures.len(), - 3, - "connection task + pump_source + one publisher for `knx://1/0/6`" - ); - } - - /// With no outbound route, `pump_sink` contributes nothing and the count - /// drops to two — the half of the contract the routed test cannot show. - #[tokio::test] - async fn an_inbound_only_db_yields_no_publisher() { - static CH: Channels<8> = Channels::new(); - let db = db().await; - let connector = KnxConnector::<_, _, 8>::new( - TokioNet::udp(Ipv4Addr::LOCALHOST), - TokioDelay, - "knx://127.0.0.1:3671", - &CH, - ); + async fn build_yields_only_the_connection_task() { + let connector = || { + KnxConnector::new( + TokioNet::udp(Ipv4Addr::LOCALHOST), + TokioDelay, + "knx://127.0.0.1:3671", + ) + }; + assert_eq!(ConnectorBuilder::scheme(&connector()), "knx"); - let futures = connector.build(&db).await.expect("build"); - assert_eq!( - futures.len(), - 2, - "connection task + pump_source, and no publisher" - ); + // With inbound and outbound routes, and with none. + for db in [routed_db().await, db().await] { + let futures = connector().build(&db).await.expect("build"); + assert_eq!(futures.len(), 1, "the connection task alone"); + } } - /// A second KNX connector is a configuration error, whether or not it - /// shares the first one's channels. - /// - /// Sharing is the louder mistake — both tasks would drain one command - /// queue — but separate channels are broken too: each connector collects - /// *all* `knx://` routes, so every `link_to` gets two publishers and - /// nothing says which gateway a route belongs to. + /// A second KNX connector is a configuration error: each connector + /// collects *all* `knx://` routes, so every route would be served twice + /// and nothing says which gateway it belongs to. #[tokio::test] async fn a_second_knx_connector_fails_the_build() { - static A: Channels<8> = Channels::new(); - static B: Channels<8> = Channels::new(); - - for (second_channels, case) in [(&A, "shared channels"), (&B, "separate channels")] { - let builder = AimDbBuilder::new() - .runtime(Arc::new(TokioAdapter)) - .with_connector(KnxConnector::<_, _, 8>::new( - TokioNet::udp(Ipv4Addr::LOCALHOST), - TokioDelay, - "knx://127.0.0.1:3671", - &A, - )) - .with_connector(KnxConnector::<_, _, 8>::new( - TokioNet::udp(Ipv4Addr::LOCALHOST), - TokioDelay, - "knx://127.0.0.2:3671", - second_channels, - )); + let builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(KnxConnector::new( + TokioNet::udp(Ipv4Addr::LOCALHOST), + TokioDelay, + "knx://127.0.0.1:3671", + )) + .with_connector(KnxConnector::new( + TokioNet::udp(Ipv4Addr::LOCALHOST), + TokioDelay, + "knx://127.0.0.2:3671", + )); - let Err(err) = builder.build().await else { - panic!("a second KNX connector must be rejected ({case})"); - }; - assert!( - format!("{err}").contains("More than one connector registered for scheme 'knx'"), - "unexpected error ({case}): {err}" - ); - } + let Err(err) = builder.build().await else { + panic!("a second KNX connector must be rejected"); + }; + assert!( + format!("{err}").contains("More than one connector registered for scheme 'knx'"), + "unexpected error: {err}" + ); } /// The whole wiring against a real UDP gateway: the task binds, advertises @@ -381,18 +235,16 @@ mod tests { /// unspecified bind takes has its own test in `client`. #[tokio::test] async fn the_wired_connector_reaches_a_gateway() { - static CH: Channels<8> = Channels::new(); let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0") .await .expect("bind gateway"); let addr = gateway.local_addr().expect("gateway addr"); let db = db().await; - let connector = KnxConnector::<_, _, 8>::new( + let connector = KnxConnector::new( TokioNet::udp(Ipv4Addr::LOCALHOST), TokioDelay, format!("knx://{addr}"), - &CH, ); let futures = connector.build(&db).await.expect("build"); let driving: Vec<_> = futures.into_iter().map(tokio::spawn).collect(); diff --git a/aimdb-knx-connector/src/lib.rs b/aimdb-knx-connector/src/lib.rs index c717b5e7..bd7f2312 100644 --- a/aimdb-knx-connector/src/lib.rs +++ b/aimdb-knx-connector/src/lib.rs @@ -17,8 +17,8 @@ //! - `std`: `connector` plus core's `std`, knx-pico's std error impls, and the //! back-compat DPT re-exports. Lifts `no_std`; adds no runtime. //! - `critical-section-std-impl`: **final binaries only** — selects -//! `critical-section`'s std impl, which `Channels` needs to link on a host. -//! An Embassy HAL already provides one. +//! `critical-section`'s std impl for a host binary that needs one. An +//! Embassy HAL already provides one. //! - `tracing`: Debug logging support (std) //! - `defmt`: Debug logging support (no_std) //! @@ -48,7 +48,7 @@ //! ```no_run //! use aimdb_core::buffer::BufferCfg; //! use aimdb_core::AimDbBuilder; -//! use aimdb_knx_connector::{Channels, KnxConnector}; +//! use aimdb_knx_connector::KnxConnector; //! use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; //! use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; //! use std::net::Ipv4Addr; @@ -62,14 +62,12 @@ //! # async fn demo() -> Result<(), Box> { //! let runtime = Arc::new(TokioAdapter::new()?); //! -//! static CHANNELS: Channels = Channels::new(); //! let mut builder = AimDbBuilder::new() //! .runtime(runtime) //! .with_connector(KnxConnector::new( //! TokioNet::udp(Ipv4Addr::UNSPECIFIED), //! TokioDelay, //! "knx://192.168.1.19:3671", -//! &CHANNELS, //! )); //! builder.configure::("light.state", |reg| { //! reg.buffer(BufferCfg::SingleLatest) @@ -100,14 +98,14 @@ //! ```rust,ignore //! use aimdb_core::AimDbBuilder; //! use aimdb_embassy_adapter::EmbassyAdapter; -//! use aimdb_knx_connector::connector::{Channels, KnxConnector}; +//! use aimdb_knx_connector::KnxConnector; //! use alloc::sync::Arc; //! //! let runtime = Arc::new(EmbassyAdapter::new()); //! //! let db = AimDbBuilder::new() //! .runtime(runtime) -//! .with_connector(KnxConnector::new(binder, EmbassyDelay, gateway, &CHANNELS)) +//! .with_connector(KnxConnector::new(binder, EmbassyDelay, gateway)) //! .configure::(|reg| { //! reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing) //! .source(sensor_producer) @@ -181,4 +179,4 @@ pub mod client; pub mod connector; #[cfg(feature = "connector")] -pub use connector::{Channels, KnxConnector}; +pub use connector::KnxConnector; diff --git a/aimdb-knx-connector/src/tunnel.rs b/aimdb-knx-connector/src/tunnel.rs index ba04cae5..d3941546 100644 --- a/aimdb-knx-connector/src/tunnel.rs +++ b/aimdb-knx-connector/src/tunnel.rs @@ -43,8 +43,8 @@ const MAX_FRAME: usize = 12 + MAX_CEMI; /// A wire datagram to send to the gateway — stack-allocated. pub type Frame = heapless::Vec; -/// Outbound GroupValueWrite command, handed to the engine by the transport's -/// command channel (fed by the `Connector::publish` side). +/// Outbound GroupValueWrite command, handed to the engine by the connection +/// task for each value it pulls from an outbound route. #[derive(Debug, Clone, PartialEq, Eq)] pub struct GroupWrite { pub group_addr: GroupAddress, @@ -79,8 +79,8 @@ pub enum Action { /// number of a tracked TUNNELING_REQUEST so a failed send can stop its /// ACK tracking (see `TunnelIo::send`); `None` for everything else. Send { frame: Frame, await_ack: Option }, - /// Deliver a parsed inbound telegram toward `pump_source` - /// (`try_send`, drop-on-full — never stall the protocol loop). + /// Deliver a parsed inbound telegram into its records (synchronous; + /// never stalls the protocol loop). Telegram { addr: GroupAddress, payload: Vec, @@ -544,8 +544,8 @@ pub(crate) trait TunnelIo { /// generic over this trait and its future has to be boxable as the /// runner's `Send` future. An impl whose socket future is `!Send` wraps it. fn send(&mut self, frame: &[u8]) -> impl core::future::Future + Send; - /// Forward a parsed telegram toward `pump_source`. Non-blocking: - /// drop + log on a full channel rather than stalling the protocol loop. + /// Forward a parsed telegram into its records. Non-blocking: never + /// stalls the protocol loop. fn forward(&mut self, addr: GroupAddress, payload: Vec); /// An outbound telegram's ACK never arrived (log-only, see /// [`Action::AckTimeout`]). diff --git a/aimdb-knx-connector/tests/shared_channel_on_std.rs b/aimdb-knx-connector/tests/shared_channel_on_std.rs deleted file mode 100644 index c4bcbd24..00000000 --- a/aimdb-knx-connector/tests/shared_channel_on_std.rs +++ /dev/null @@ -1,53 +0,0 @@ -//! The MCU's channel and select types must work in a **linked** std binary, so -//! one connection task can serve both runtimes. -//! -//! `CriticalSectionRawMutex` is the only `Sync` raw mutex `embassy-sync` offers -//! — `NoopRawMutex` is `!Sync` and cannot back a shared channel at all — and -//! using it pulls in `_critical_section_1_0_acquire`/`_release`, which nothing -//! defines on std. Selecting an impl is the final binary's call, so the library -//! does not make it: a test binary is a binary, and gets the std impl through -//! this crate's `critical-section` dev-dependency. These tests fail to *link*, -//! not to compile, if that ever comes undone. -#![cfg(feature = "std")] - -use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; -use embassy_sync::channel::Channel; - -type Cmd = (u8, u16); - -/// A send and receive through the channel, exercising the critical section. -#[tokio::test] -async fn embassy_channel_carries_a_value_on_std() { - let channel: Channel = Channel::new(); - - channel.send((1, 0x0a0b)).await; - assert_eq!(channel.receive().await, (1, 0x0a0b)); -} - -/// The channel is `Send + Sync`, so a spawned task can enqueue while the -/// protocol loop drains — what the unified connection task relies on. -#[tokio::test] -async fn embassy_channel_is_shareable_across_tasks_on_std() { - static CHANNEL: Channel = Channel::new(); - - let producer = tokio::spawn(async { - CHANNEL.send((2, 0x0c0d)).await; - }); - - assert_eq!(CHANNEL.receive().await, (2, 0x0c0d)); - producer.await.expect("producer task"); -} - -/// `embassy-futures` drives the same select on std as on the MCU. -#[tokio::test] -async fn embassy_select_resolves_on_std() { - use embassy_futures::select::{select, Either}; - - let channel: Channel = Channel::new(); - channel.send((3, 0x0e0f)).await; - - match select(channel.receive(), core::future::pending::<()>()).await { - Either::First(cmd) => assert_eq!(cmd, (3, 0x0e0f)), - Either::Second(()) => panic!("pending future must never win"), - } -} diff --git a/aimdb-knx-connector/tests/topic_provider_tests.rs b/aimdb-knx-connector/tests/topic_provider_tests.rs deleted file mode 100644 index 2441dfa7..00000000 --- a/aimdb-knx-connector/tests/topic_provider_tests.rs +++ /dev/null @@ -1,590 +0,0 @@ -//! Integration tests for KNX TopicProvider and TopicResolver functionality -//! -//! These tests verify the dynamic group address routing features: -//! - **TopicProvider**: Outbound (AimDB → KNX) dynamic group address selection per-value -//! - **TopicResolver**: Inbound (KNX → AimDB) late-binding group address resolution at startup -//! -//! The tests use mock data and don't require a running KNX/IP gateway. - -#![cfg(feature = "std")] - -use aimdb_core::buffer::BufferCfg; -use aimdb_core::connector::TopicProvider; -use aimdb_core::{AimDbBuilder, Producer, RuntimeContext}; -use aimdb_knx_connector::Channels; -use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; -use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; -use std::net::Ipv4Addr; -use std::sync::atomic::{AtomicU32, Ordering}; -use std::sync::Arc; - -/// The connector's channel pair, shared by the registration tests — none of -/// them runs a connection task, so one pair serves all three. -static CHANNELS: Channels = Channels::new(); - -// ============================================================================ -// Test Types -// ============================================================================ - -/// Dimmer value with room identifier for dynamic group address routing -#[derive(Clone, Debug)] -struct DimmerValue { - room_id: String, - level: u8, // 0-255 -} - -impl DimmerValue { - fn new(room_id: &str, level: u8) -> Self { - Self { - room_id: room_id.into(), - level, - } - } - - fn to_knx_bytes(&self) -> Vec { - vec![self.level] - } -} - -/// Switch state with zone identifier -#[derive(Clone, Debug)] -struct SwitchState { - zone_id: String, - is_on: bool, -} - -impl SwitchState { - fn new(zone_id: &str, is_on: bool) -> Self { - Self { - zone_id: zone_id.into(), - is_on, - } - } - - #[allow(dead_code)] - fn to_knx_bytes(&self) -> Vec { - vec![if self.is_on { 1 } else { 0 }] - } - - fn from_knx_bytes(data: &[u8]) -> Result { - if data.is_empty() { - return Err("Empty data".into()); - } - Ok(SwitchState { - zone_id: "unknown".into(), - is_on: data[0] != 0, - }) - } -} - -/// Temperature setpoint for HVAC control -#[derive(Clone, Debug)] -struct TemperatureSetpoint { - hvac_zone: u8, // Zone number 1-16 - setpoint: f32, // Temperature in Celsius -} - -impl TemperatureSetpoint { - fn new(hvac_zone: u8, setpoint: f32) -> Self { - Self { - hvac_zone, - setpoint, - } - } - - fn to_knx_bytes(&self) -> Vec { - // KNX DPT 9.001 encoding (simplified for test) - let value = (self.setpoint * 100.0) as i16; - vec![(value >> 8) as u8, value as u8] - } -} - -// ============================================================================ -// TopicProvider Implementations (KNX Group Addresses) -// ============================================================================ - -/// Dynamic group address provider based on room ID -/// -/// Routes dimmer commands to different KNX group addresses based on room. -/// In a real building, each room has its own dimmer actuator address. -struct RoomBasedGroupAddressProvider { - /// Base group address (main/middle) - base_main: u8, - base_middle: u8, -} - -impl RoomBasedGroupAddressProvider { - fn new(base_main: u8, base_middle: u8) -> Self { - Self { - base_main, - base_middle, - } - } - - fn room_to_sub(&self, room_id: &str) -> u8 { - // Map room names to sub-addresses - match room_id { - "living" => 1, - "bedroom" => 2, - "kitchen" => 3, - "bathroom" => 4, - "office" => 5, - _ => 0, // Default/unknown - } - } -} - -impl TopicProvider for RoomBasedGroupAddressProvider { - fn topic(&self, value: &DimmerValue) -> Option { - let sub = self.room_to_sub(&value.room_id); - Some(format!( - "knx://{}/{}/{}", - self.base_main, self.base_middle, sub - )) - } -} - -/// HVAC zone-based group address provider -/// -/// Routes temperature setpoints to zone-specific group addresses. -struct HvacZoneProvider; - -impl TopicProvider for HvacZoneProvider { - fn topic(&self, value: &TemperatureSetpoint) -> Option { - // HVAC zones mapped to group addresses 5/0/1 through 5/0/16 - if value.hvac_zone >= 1 && value.hvac_zone <= 16 { - Some(format!("knx://5/0/{}", value.hvac_zone)) - } else { - None // Invalid zone, use fallback - } - } -} - -/// Switch provider with emergency override -/// -/// Demonstrates conditional routing: emergency signals go to broadcast address. -struct SwitchWithEmergencyProvider; - -impl TopicProvider for SwitchWithEmergencyProvider { - fn topic(&self, value: &SwitchState) -> Option { - if value.zone_id == "emergency" { - // Emergency signals go to broadcast group - Some("knx://0/0/0".into()) - } else if value.zone_id.starts_with("zone-") { - // Parse zone number - let zone_num: u8 = value.zone_id[5..].parse().unwrap_or(0); - Some(format!("knx://1/1/{}", zone_num)) - } else { - None // Use default from link_to() - } - } -} - -// ============================================================================ -// Unit Tests for KNX TopicProvider -// ============================================================================ - -#[test] -fn test_room_based_group_address_provider() { - let provider = RoomBasedGroupAddressProvider::new(1, 0); - - let living = DimmerValue::new("living", 128); - assert_eq!(provider.topic(&living), Some("knx://1/0/1".into())); - - let bedroom = DimmerValue::new("bedroom", 255); - assert_eq!(provider.topic(&bedroom), Some("knx://1/0/2".into())); - - let kitchen = DimmerValue::new("kitchen", 64); - assert_eq!(provider.topic(&kitchen), Some("knx://1/0/3".into())); - - // Unknown room uses sub-address 0 - let unknown = DimmerValue::new("garage", 100); - assert_eq!(provider.topic(&unknown), Some("knx://1/0/0".into())); -} - -#[test] -fn test_hvac_zone_provider() { - let provider = HvacZoneProvider; - - let zone1 = TemperatureSetpoint::new(1, 21.0); - assert_eq!(provider.topic(&zone1), Some("knx://5/0/1".into())); - - let zone16 = TemperatureSetpoint::new(16, 22.5); - assert_eq!(provider.topic(&zone16), Some("knx://5/0/16".into())); - - // Invalid zone returns None (fallback) - let invalid = TemperatureSetpoint::new(0, 20.0); - assert_eq!(provider.topic(&invalid), None); - - let too_high = TemperatureSetpoint::new(17, 20.0); - assert_eq!(provider.topic(&too_high), None); -} - -#[test] -fn test_switch_with_emergency_provider() { - let provider = SwitchWithEmergencyProvider; - - // Emergency goes to broadcast - let emergency = SwitchState::new("emergency", true); - assert_eq!(provider.topic(&emergency), Some("knx://0/0/0".into())); - - // Normal zones - let zone1 = SwitchState::new("zone-1", true); - assert_eq!(provider.topic(&zone1), Some("knx://1/1/1".into())); - - let zone5 = SwitchState::new("zone-5", false); - assert_eq!(provider.topic(&zone5), Some("knx://1/1/5".into())); - - // Unknown zone uses fallback - let unknown = SwitchState::new("lobby", true); - assert_eq!(provider.topic(&unknown), None); -} - -// ============================================================================ -// Unit Tests for KNX TopicResolver -// ============================================================================ - -#[test] -fn test_topic_resolver_from_config_file() { - // Simulate reading group address from a config file or service - std::env::set_var("KNX_SWITCH_GROUP", "1/2/3"); - - let resolver = || { - std::env::var("KNX_SWITCH_GROUP") - .ok() - .map(|addr| format!("knx://{}", addr)) - }; - - assert_eq!(resolver(), Some("knx://1/2/3".into())); - - std::env::remove_var("KNX_SWITCH_GROUP"); -} - -#[test] -fn test_topic_resolver_with_discovery() { - use std::sync::Mutex; - - // Simulate KNX device discovery - let discovered_devices = Arc::new(Mutex::new(vec![ - ("dimmer-1", "1/0/1"), - ("dimmer-2", "1/0/2"), - ])); - - let devices_clone = discovered_devices.clone(); - let resolver = move || { - devices_clone - .lock() - .unwrap() - .iter() - .find(|(name, _)| *name == "dimmer-1") - .map(|(_, addr)| format!("knx://{}", addr)) - }; - - assert_eq!(resolver(), Some("knx://1/0/1".into())); -} - -// ============================================================================ -// Integration Test: TopicProvider with AimDbBuilder -// ============================================================================ - -/// Test that TopicProvider can be configured without connector (verifies API) -#[tokio::test] -async fn test_knx_topic_provider_registration_api() { - let runtime = Arc::new(TokioAdapter::new().unwrap()); - let produced_count = Arc::new(AtomicU32::new(0)); - let produced_count_clone = produced_count.clone(); - - let mut builder = AimDbBuilder::new().runtime(runtime); - - // Register without link_to (no connector needed) - builder.configure::("knx.dimmer.living", |reg| { - let counter = produced_count_clone.clone(); - reg.buffer(BufferCfg::SingleLatest).source( - move |_ctx: RuntimeContext, producer: Producer| { - let counter = counter.clone(); - async move { - let dimmer = DimmerValue::new("living", 200); - producer.produce(dimmer); - counter.fetch_add(1, Ordering::SeqCst); - } - }, - ); - }); - - // Build should succeed - assert!(builder.build().await.is_ok()); -} - -/// Test TopicProvider with KnxConnector registration -/// -/// This test verifies the full configuration API including link_to + with_topic_provider -/// works correctly at compile time. Runtime requires actual KNX gateway. -#[tokio::test] -async fn test_knx_topic_provider_with_connector_registration() { - let runtime = Arc::new(TokioAdapter::new().unwrap()); - - let mut builder = AimDbBuilder::new().runtime(runtime).with_connector( - aimdb_knx_connector::KnxConnector::new( - TokioNet::udp(Ipv4Addr::UNSPECIFIED), - TokioDelay, - "knx://192.168.1.10:3671", - &CHANNELS, - ), - ); - - // Register dimmer with dynamic group address provider - builder.configure::("knx.dimmer.living", |reg| { - reg.buffer(BufferCfg::SingleLatest) - .source( - |_ctx: RuntimeContext, producer: Producer| async move { - let dimmer = DimmerValue::new("living", 200); - producer.produce(dimmer); - }, - ) - .link_to("knx://1/0/0") // Fallback group address - .with_topic_provider(RoomBasedGroupAddressProvider::new(1, 0)) - .with_serializer(|_ctx, dimmer: &DimmerValue| Ok(dimmer.to_knx_bytes())) - .finish(); - }); - - // Build succeeds with connector registered - assert!(builder.build().await.is_ok()); -} - -#[tokio::test] -async fn test_knx_topic_resolver_with_connector_registration() { - let runtime = Arc::new(TokioAdapter::new().unwrap()); - - std::env::set_var("KNX_SWITCH_INPUT", "1/2/10"); - - let mut builder = AimDbBuilder::new().runtime(runtime).with_connector( - aimdb_knx_connector::KnxConnector::new( - TokioNet::udp(Ipv4Addr::UNSPECIFIED), - TokioDelay, - "knx://192.168.1.10:3671", - &CHANNELS, - ), - ); - - // Register switch with dynamic group address resolver - builder.configure::("knx.switch.zone1", |reg| { - reg.buffer(BufferCfg::SingleLatest) - .link_from("knx://1/2/0") // Fallback group address - .with_topic_resolver(|| { - std::env::var("KNX_SWITCH_INPUT") - .ok() - .map(|addr| format!("knx://{}", addr)) - }) - .with_deserializer(|_ctx, data: &[u8]| SwitchState::from_knx_bytes(data)) - .finish(); - }); - - // Build should succeed - assert!(builder.build().await.is_ok()); - - std::env::remove_var("KNX_SWITCH_INPUT"); -} - -#[tokio::test] -async fn test_hvac_zone_routing() { - let runtime = Arc::new(TokioAdapter::new().unwrap()); - - let mut builder = AimDbBuilder::new().runtime(runtime).with_connector( - aimdb_knx_connector::KnxConnector::new( - TokioNet::udp(Ipv4Addr::UNSPECIFIED), - TokioDelay, - "knx://192.168.1.10:3671", - &CHANNELS, - ), - ); - - // HVAC setpoint with zone-based routing - builder.configure::("knx.hvac.setpoint", |reg| { - reg.buffer(BufferCfg::SingleLatest) - .source( - |_ctx: RuntimeContext, producer: Producer| async move { - // Different zones get routed to different group addresses - for zone in 1..=4 { - let setpoint = TemperatureSetpoint::new(zone, 21.0 + zone as f32 * 0.5); - producer.produce(setpoint); - } - }, - ) - .link_to("knx://5/0/0") // Fallback for invalid zones - .with_topic_provider(HvacZoneProvider) - .with_serializer(|_ctx, sp: &TemperatureSetpoint| Ok(sp.to_knx_bytes())) - .finish(); - }); - - assert!(builder.build().await.is_ok()); -} - -// ============================================================================ -// Test TopicProvider as a typed trait object with KNX Types -// ============================================================================ - -#[test] -fn test_knx_topic_provider_as_trait_object() { - use aimdb_core::connector::TopicProvider; - use std::sync::Arc; - - // Providers are stored as Arc> and stay typed - // end-to-end — a wrong-type call is unrepresentable. - let provider: Arc> = - Arc::new(RoomBasedGroupAddressProvider::new(1, 0)); - - let dimmer = DimmerValue::new("living", 128); - assert_eq!(provider.topic(&dimmer), Some("knx://1/0/1".into())); -} - -// ============================================================================ -// Test: Simulate Connector Group Address Resolution Logic -// ============================================================================ -// -// These tests simulate EXACTLY what the fused outbound reader does internally -// while it still holds the typed value: -// ```rust -// let dest = topic.as_ref().and_then(|p| p.topic(&value)); -// // ...later, in the pump: -// let dest = msg.dest.unwrap_or_else(|| default_group_addr.clone()); -// ``` - -/// Simulates the fused reader's group address resolution for outbound telegrams -fn resolve_group_address_like_connector( - default_group_addr: &str, - topic_provider: Option<&dyn aimdb_core::connector::TopicProvider>, - value: &T, -) -> String { - topic_provider - .and_then(|provider| provider.topic(value)) - .unwrap_or_else(|| default_group_addr.to_string()) -} - -#[test] -fn test_connector_group_address_resolution_room_based() { - let provider = RoomBasedGroupAddressProvider::new(1, 0); - let default_addr = "knx://1/0/0"; - - // Test 1: Living room dimmer - let dimmer_living = DimmerValue::new("living", 128); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_living); - assert_eq!(resolved, "knx://1/0/1"); - - // Test 2: Bedroom dimmer - let dimmer_bedroom = DimmerValue::new("bedroom", 255); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_bedroom); - assert_eq!(resolved, "knx://1/0/2"); - - // Test 3: Kitchen dimmer - let dimmer_kitchen = DimmerValue::new("kitchen", 64); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_kitchen); - assert_eq!(resolved, "knx://1/0/3"); - - // Test 4: Unknown room → sub-address 0 - let dimmer_unknown = DimmerValue::new("garage", 100); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &dimmer_unknown); - assert_eq!(resolved, "knx://1/0/0"); -} - -#[test] -fn test_connector_group_address_resolution_hvac_zones() { - let provider = HvacZoneProvider; - let default_addr = "knx://5/0/0"; // Fallback for invalid zones - - // Test valid zones 1-16 - for zone in 1..=16u8 { - let setpoint = TemperatureSetpoint::new(zone, 21.0); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &setpoint); - assert_eq!(resolved, format!("knx://5/0/{}", zone)); - } - - // Test invalid zone 0 → fallback - let invalid_zone_0 = TemperatureSetpoint::new(0, 21.0); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &invalid_zone_0); - assert_eq!(resolved, "knx://5/0/0"); // Fallback - - // Test invalid zone 17 → fallback - let invalid_zone_17 = TemperatureSetpoint::new(17, 21.0); - let resolved = - resolve_group_address_like_connector(default_addr, Some(&provider), &invalid_zone_17); - assert_eq!(resolved, "knx://5/0/0"); // Fallback -} - -#[test] -fn test_connector_group_address_resolution_emergency_switch() { - let provider = SwitchWithEmergencyProvider; - let default_addr = "knx://1/1/0"; - - // Test 1: Emergency switch → broadcast address - let emergency = SwitchState::new("emergency", true); - let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &emergency); - assert_eq!(resolved, "knx://0/0/0"); // Broadcast! - - // Test 2: Normal zone switches - let zone_1 = SwitchState::new("zone-1", true); - let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &zone_1); - assert_eq!(resolved, "knx://1/1/1"); - - let zone_5 = SwitchState::new("zone-5", false); - let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &zone_5); - assert_eq!(resolved, "knx://1/1/5"); - - // Test 3: Unknown zone → fallback - let lobby = SwitchState::new("lobby", true); - let resolved = resolve_group_address_like_connector(default_addr, Some(&provider), &lobby); - assert_eq!(resolved, "knx://1/1/0"); // Fallback -} - -#[test] -fn test_connector_group_address_no_provider() { - let default_addr = "knx://1/0/99"; - - // No provider → always use default - let dimmer = DimmerValue::new("living", 128); - let resolved = resolve_group_address_like_connector(default_addr, None, &dimmer); - assert_eq!(resolved, "knx://1/0/99"); -} - -/// Test inbound group address resolution (TopicResolver) -#[test] -fn test_inbound_group_address_resolver_simulation() { - fn resolve_inbound_group_addr( - default_addr: &str, - resolver: Option<&dyn Fn() -> Option>, - ) -> String { - resolver - .and_then(|r| r()) - .unwrap_or_else(|| default_addr.to_string()) - } - - // Test 1: Resolver returns dynamic address - let resolver_dynamic = || Some("knx://1/2/10".to_string()); - let addr = resolve_inbound_group_addr("knx://1/2/0", Some(&resolver_dynamic)); - assert_eq!(addr, "knx://1/2/10"); - - // Test 2: Resolver returns None → fallback - let resolver_fallback = || None; - let addr = resolve_inbound_group_addr("knx://1/2/0", Some(&resolver_fallback)); - assert_eq!(addr, "knx://1/2/0"); - - // Test 3: No resolver → default - let addr = resolve_inbound_group_addr("knx://1/2/0", None); - assert_eq!(addr, "knx://1/2/0"); - - // Test 4: Resolver from environment (simulating config file) - std::env::set_var("KNX_INBOUND_ADDR", "1/3/5"); - let resolver_env = || { - std::env::var("KNX_INBOUND_ADDR") - .ok() - .map(|a| format!("knx://{}", a)) - }; - let addr = resolve_inbound_group_addr("knx://1/0/0", Some(&resolver_env)); - assert_eq!(addr, "knx://1/3/5"); - std::env::remove_var("KNX_INBOUND_ADDR"); -} diff --git a/aimdb-knx-connector/tests/topic_writer_tests.rs b/aimdb-knx-connector/tests/topic_writer_tests.rs new file mode 100644 index 00000000..42055a0b --- /dev/null +++ b/aimdb-knx-connector/tests/topic_writer_tests.rs @@ -0,0 +1,405 @@ +//! Integration tests for KNX TopicWriter and TopicResolver functionality +//! +//! These tests verify the dynamic group address routing features: +//! - **TopicWriter**: Outbound (AimDB → KNX) dynamic group address selection per-value +//! - **TopicResolver**: Inbound (KNX → AimDB) late-binding group address resolution at startup +//! +//! A writer writes the bare group address (`1/0/1`), not the `knx://` URL. The +//! longest group address, `31/7/255`, is 8 bytes. +//! +//! The tests use mock data and don't require a running KNX/IP gateway. + +#![cfg(feature = "std")] + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::connector::{TopicBuf, TopicOverflow, TopicWriter}; +use aimdb_core::{AimDbBuilder, Producer, RuntimeContext}; +use aimdb_knx_connector::KnxConnector; +use aimdb_tokio_adapter::net::{TokioDelay, TokioNet, TokioUdpBinder}; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use std::fmt::Write as _; +use std::net::Ipv4Addr; +use std::sync::Arc; + +/// Topic capacity for every KNX writer here: the longest group address. +const GROUP_ADDRESS_CAPACITY: usize = 8; + +// ============================================================================ +// Test Types +// ============================================================================ + +/// Dimmer value with room identifier for dynamic group address routing +#[derive(Clone, Debug)] +struct DimmerValue { + room_id: String, + level: u8, // 0-255 +} + +impl DimmerValue { + fn new(room_id: &str, level: u8) -> Self { + Self { + room_id: room_id.into(), + level, + } + } + + fn to_knx_bytes(&self) -> Vec { + vec![self.level] + } +} + +/// Switch state with zone identifier +#[derive(Clone, Debug)] +struct SwitchState { + zone_id: String, + #[allow(dead_code)] + is_on: bool, +} + +impl SwitchState { + fn new(zone_id: &str, is_on: bool) -> Self { + Self { + zone_id: zone_id.into(), + is_on, + } + } + + fn from_knx_bytes(data: &[u8]) -> Result { + if data.is_empty() { + return Err("Empty data".into()); + } + Ok(SwitchState { + zone_id: "unknown".into(), + is_on: data[0] != 0, + }) + } +} + +/// Temperature setpoint for HVAC control +#[derive(Clone, Debug)] +struct TemperatureSetpoint { + hvac_zone: u8, // Zone number 1-16 + setpoint: f32, // Temperature in Celsius +} + +impl TemperatureSetpoint { + fn new(hvac_zone: u8, setpoint: f32) -> Self { + Self { + hvac_zone, + setpoint, + } + } + + fn to_knx_bytes(&self) -> Vec { + // KNX DPT 9.001 encoding (simplified for test) + let value = (self.setpoint * 100.0) as i16; + vec![(value >> 8) as u8, value as u8] + } +} + +// ============================================================================ +// TopicWriter Implementations (KNX Group Addresses) +// ============================================================================ + +/// Dynamic group address writer based on room ID +/// +/// Routes dimmer commands to different KNX group addresses based on room. +/// In a real building, each room has its own dimmer actuator address. +struct RoomBasedGroupAddress { + /// Base group address (main/middle) + base_main: u8, + base_middle: u8, +} + +impl RoomBasedGroupAddress { + fn new(base_main: u8, base_middle: u8) -> Self { + Self { + base_main, + base_middle, + } + } + + fn room_to_sub(&self, room_id: &str) -> u8 { + // Map room names to sub-addresses + match room_id { + "living" => 1, + "bedroom" => 2, + "kitchen" => 3, + "bathroom" => 4, + "office" => 5, + _ => 0, // Default/unknown + } + } +} + +impl TopicWriter for RoomBasedGroupAddress { + fn write_topic( + &self, + value: &DimmerValue, + out: &mut TopicBuf<'_>, + ) -> Result { + let sub = self.room_to_sub(&value.room_id); + write!(out, "{}/{}/{}", self.base_main, self.base_middle, sub)?; + Ok(true) + } +} + +/// HVAC zone-based group address writer +/// +/// Routes temperature setpoints to zone-specific group addresses. +struct HvacZone; + +impl TopicWriter for HvacZone { + fn write_topic( + &self, + value: &TemperatureSetpoint, + out: &mut TopicBuf<'_>, + ) -> Result { + // HVAC zones mapped to group addresses 5/0/1 through 5/0/16 + if !(1..=16).contains(&value.hvac_zone) { + return Ok(false); // Invalid zone, use fallback + } + write!(out, "5/0/{}", value.hvac_zone)?; + Ok(true) + } +} + +/// Switch writer with emergency override +/// +/// Demonstrates conditional routing: emergency signals go to broadcast address. +struct SwitchWithEmergency; + +impl TopicWriter for SwitchWithEmergency { + fn write_topic( + &self, + value: &SwitchState, + out: &mut TopicBuf<'_>, + ) -> Result { + if value.zone_id == "emergency" { + // Emergency signals go to broadcast group + out.push_str("0/0/0")?; + } else if let Some(zone) = value.zone_id.strip_prefix("zone-") { + let zone_num: u8 = zone.parse().unwrap_or(0); + write!(out, "1/1/{zone_num}")?; + } else { + return Ok(false); // Use default from link_to() + } + Ok(true) + } +} + +/// The group address `writer` selects for `value`, or `default` when it +/// returns `Ok(false)` — what the outbound route resolves. +fn resolve(writer: &dyn TopicWriter, value: &T, default: &str) -> String { + let mut storage = [0u8; GROUP_ADDRESS_CAPACITY]; + let mut out = TopicBuf::new(&mut storage); + match writer.write_topic(value, &mut out) { + Ok(true) => out.as_str().to_string(), + Ok(false) => default.to_string(), + Err(TopicOverflow) => panic!("group address exceeds {GROUP_ADDRESS_CAPACITY} bytes"), + } +} + +// ============================================================================ +// Unit Tests for KNX TopicWriter +// ============================================================================ + +#[test] +fn test_room_based_group_address() { + let writer = RoomBasedGroupAddress::new(1, 0); + let default = "1/0/0"; + + for (room, expected) in [ + ("living", "1/0/1"), + ("bedroom", "1/0/2"), + ("kitchen", "1/0/3"), + // Unknown room uses sub-address 0 + ("garage", "1/0/0"), + ] { + let dimmer = DimmerValue::new(room, 128); + assert_eq!(resolve(&writer, &dimmer, default), expected, "{room}"); + } +} + +#[test] +fn test_hvac_zone_routing_and_fallback() { + let default = "5/0/0"; // Fallback for invalid zones + + for zone in 1..=16u8 { + let setpoint = TemperatureSetpoint::new(zone, 21.0); + assert_eq!( + resolve(&HvacZone, &setpoint, default), + format!("5/0/{zone}") + ); + } + + // Invalid zones fall back to the link's default. + for zone in [0, 17] { + let setpoint = TemperatureSetpoint::new(zone, 20.0); + assert_eq!(resolve(&HvacZone, &setpoint, default), default); + } +} + +#[test] +fn test_switch_with_emergency() { + let default = "1/1/0"; + + // Emergency goes to broadcast + let emergency = SwitchState::new("emergency", true); + assert_eq!(resolve(&SwitchWithEmergency, &emergency, default), "0/0/0"); + + // Normal zones + let zone1 = SwitchState::new("zone-1", true); + assert_eq!(resolve(&SwitchWithEmergency, &zone1, default), "1/1/1"); + + let zone5 = SwitchState::new("zone-5", false); + assert_eq!(resolve(&SwitchWithEmergency, &zone5, default), "1/1/5"); + + // Unknown zone uses fallback + let lobby = SwitchState::new("lobby", true); + assert_eq!(resolve(&SwitchWithEmergency, &lobby, default), default); +} + +/// The longest group address fits the capacity; one byte less overflows. +#[test] +fn test_longest_group_address_fits_the_capacity() { + let writer = RoomBasedGroupAddress::new(31, 7); + let dimmer = DimmerValue::new("living", 1); + + let mut storage = [0u8; GROUP_ADDRESS_CAPACITY]; + let mut out = TopicBuf::new(&mut storage); + assert_eq!(writer.write_topic(&dimmer, &mut out), Ok(true)); + assert_eq!(out.as_str(), "31/7/1"); + + let writer = |out: &mut TopicBuf<'_>| write!(out, "31/7/255"); + let mut storage = [0u8; GROUP_ADDRESS_CAPACITY]; + assert!(writer(&mut TopicBuf::new(&mut storage)).is_ok()); + let mut storage = [0u8; GROUP_ADDRESS_CAPACITY - 1]; + assert!(writer(&mut TopicBuf::new(&mut storage)).is_err()); +} + +// ============================================================================ +// Unit Tests for KNX TopicResolver +// ============================================================================ + +#[test] +fn test_topic_resolver_from_config_file() { + // Simulate reading group address from a config file or service + std::env::set_var("KNX_SWITCH_GROUP", "1/2/3"); + + let resolver = || { + std::env::var("KNX_SWITCH_GROUP") + .ok() + .map(|addr| format!("knx://{}", addr)) + }; + + assert_eq!(resolver(), Some("knx://1/2/3".into())); + + std::env::remove_var("KNX_SWITCH_GROUP"); +} + +// ============================================================================ +// Integration Tests: registration with KnxConnector +// ============================================================================ + +fn connector() -> KnxConnector { + KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, + "knx://192.168.1.10:3671", + ) +} + +/// `link_to` + `with_topic_writer` builds with the connector registered. +/// Running it would need a KNX gateway. +#[tokio::test] +async fn test_knx_topic_writer_with_connector_registration() { + let runtime = Arc::new(TokioAdapter::new().unwrap()); + let mut builder = AimDbBuilder::new() + .runtime(runtime) + .with_connector(connector()); + + builder.configure::("knx.dimmer.living", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .source( + |_ctx: RuntimeContext, producer: Producer| async move { + producer.produce(DimmerValue::new("living", 200)); + }, + ) + .link_to("knx://1/0/0") // Fallback group address + .with_topic_writer(GROUP_ADDRESS_CAPACITY, RoomBasedGroupAddress::new(1, 0)) + .with_serializer(|_ctx, dimmer: &DimmerValue| Ok(dimmer.to_knx_bytes())) + .finish(); + }); + builder.configure::("knx.hvac.setpoint", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_to("knx://5/0/0") // Fallback for invalid zones + .with_topic_writer(GROUP_ADDRESS_CAPACITY, HvacZone) + .with_serializer(|_ctx, sp: &TemperatureSetpoint| Ok(sp.to_knx_bytes())) + .finish(); + }); + + assert!(builder.build().await.is_ok()); +} + +/// The connector pulls through `OutboundRoutes`, which rejects +/// `with_topic_provider` links at build. +#[tokio::test] +async fn test_knx_rejects_a_topic_provider() { + use aimdb_core::connector::TopicProvider; + + struct Fixed; + impl TopicProvider for Fixed { + fn topic(&self, _value: &DimmerValue) -> Option { + Some("1/0/1".into()) + } + } + + let runtime = Arc::new(TokioAdapter::new().unwrap()); + let mut builder = AimDbBuilder::new() + .runtime(runtime) + .with_connector(connector()); + builder.configure::("knx.dimmer.living", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_to("knx://1/0/0") + .with_topic_provider(Fixed) + .with_serializer(|_ctx, dimmer: &DimmerValue| Ok(dimmer.to_knx_bytes())) + .finish(); + }); + + let Err(err) = builder.build().await else { + panic!("a topic provider must be rejected"); + }; + assert!( + format!("{err}").contains("with_topic_provider"), + "unexpected error: {err}" + ); +} + +#[tokio::test] +async fn test_knx_topic_resolver_with_connector_registration() { + let runtime = Arc::new(TokioAdapter::new().unwrap()); + + std::env::set_var("KNX_SWITCH_INPUT", "1/2/10"); + + let mut builder = AimDbBuilder::new() + .runtime(runtime) + .with_connector(connector()); + + // Register switch with dynamic group address resolver + builder.configure::("knx.switch.zone1", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_from("knx://1/2/0") // Fallback group address + .with_topic_resolver(|| { + std::env::var("KNX_SWITCH_INPUT") + .ok() + .map(|addr| format!("knx://{}", addr)) + }) + .with_deserializer(|_ctx, data: &[u8]| SwitchState::from_knx_bytes(data)) + .finish(); + }); + + assert!(builder.build().await.is_ok()); + + std::env::remove_var("KNX_SWITCH_INPUT"); +} diff --git a/docs/aimdb-usage-guide.md b/docs/aimdb-usage-guide.md index f5dee285..9cdfdb0e 100644 --- a/docs/aimdb-usage-guide.md +++ b/docs/aimdb-usage-guide.md @@ -315,7 +315,7 @@ knx-pico = { git = "https://github.com/aimdb-dev/knx-pico.git", branch = "master use aimdb_core::prelude::*; use aimdb_tokio_adapter::TokioAdapter; use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; -use aimdb_knx_connector::{Channels, KnxConnector}; +use aimdb_knx_connector::KnxConnector; use std::net::Ipv4Addr; use std::sync::Arc; @@ -324,9 +324,6 @@ struct LightState { is_on: bool, } -// One channel pair per connector, held for the process lifetime. -static CHANNELS: Channels = Channels::new(); - #[tokio::main] async fn main() -> Result<(), Box> { let runtime = Arc::new(TokioAdapter::new()?); @@ -339,7 +336,6 @@ async fn main() -> Result<(), Box> { TokioNet::udp(Ipv4Addr::UNSPECIFIED), TokioDelay, "knx://192.168.1.19:3671", - &CHANNELS, )) .configure::("light.state", |reg| { reg.buffer(BufferCfg::SingleLatest) diff --git a/examples/embassy-knx-connector-demo/src/main.rs b/examples/embassy-knx-connector-demo/src/main.rs index 7fd463cc..52286d37 100644 --- a/examples/embassy-knx-connector-demo/src/main.rs +++ b/examples/embassy-knx-connector-demo/src/main.rs @@ -45,7 +45,7 @@ use aimdb_core::{AimDbBuilder, RecordKey, RuntimeContext}; use aimdb_embassy_adapter::io::EmbassyUart; use aimdb_embassy_adapter::net::{EmbassyDelay, EmbassyNet}; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; -use aimdb_knx_connector::connector::{Channels, KnxConnector}; +use aimdb_knx_connector::connector::KnxConnector; use aimdb_knx_connector::dpt::{Dpt1, Dpt9, DptDecode, DptEncode}; use aimdb_serial_connector::SerialServer; use defmt::*; @@ -273,12 +273,11 @@ async fn main(spawner: Spawner) { let (serial_tx, serial_rx) = uart.split(); // The adapter owns the UDP socket and the clock; the connector owns the - // tunnelling protocol. Buffers and channels are `'static`, as on any MCU. + // tunnelling protocol. Socket buffers are `'static`, as on any MCU. static KNX_RX_META: StaticCell<[PacketMetadata; 8]> = StaticCell::new(); static KNX_RX_BUF: StaticCell<[u8; 1024]> = StaticCell::new(); static KNX_TX_META: StaticCell<[PacketMetadata; 8]> = StaticCell::new(); static KNX_TX_BUF: StaticCell<[u8; 1024]> = StaticCell::new(); - static KNX_CHANNELS: Channels<32> = Channels::new(); let knx_binder = EmbassyNet::udp( *stack, KNX_RX_META.init([PacketMetadata::EMPTY; 8]), @@ -291,12 +290,7 @@ async fn main(spawner: Spawner) { // remote `record.set` is refused — peers can list/get/subscribe, not write. let mut builder = AimDbBuilder::new() .runtime(runtime.clone()) - .with_connector(KnxConnector::new( - knx_binder, - EmbassyDelay, - &gateway_url, - &KNX_CHANNELS, - )) + .with_connector(KnxConnector::new(knx_binder, EmbassyDelay, &gateway_url)) .with_connector( SerialServer::new(EmbassyUart::new(serial_rx, serial_tx)) .security_policy(SecurityPolicy::read_only()), diff --git a/examples/tokio-knx-connector-demo/README.md b/examples/tokio-knx-connector-demo/README.md index 3f5460c1..328ebc33 100644 --- a/examples/tokio-knx-connector-demo/README.md +++ b/examples/tokio-knx-connector-demo/README.md @@ -27,16 +27,12 @@ Demonstrates bidirectional KNX/IP integration with AimDB using the Tokio runtime Edit `src/main.rs` to match your KNX setup: ```rust -// The connector's queues, held for the process lifetime. -static KNX_CHANNELS: Channels = Channels::new(); - // The adapter owns the UDP socket and the clock; the connector owns the // tunnelling protocol. Only the gateway URL needs changing. .with_connector(KnxConnector::new( TokioNet::udp(Ipv4Addr::UNSPECIFIED), TokioDelay, "knx://YOUR_GATEWAY_IP:3671", // Change to your gateway IP - &KNX_CHANNELS, )) // Group addresses diff --git a/examples/tokio-knx-connector-demo/src/main.rs b/examples/tokio-knx-connector-demo/src/main.rs index fe266d9b..8fc88ef3 100644 --- a/examples/tokio-knx-connector-demo/src/main.rs +++ b/examples/tokio-knx-connector-demo/src/main.rs @@ -25,7 +25,7 @@ use aimdb_core::buffer::BufferCfg; use aimdb_core::remote::{AimxConfig, SecurityPolicy}; use aimdb_core::{AimDbBuilder, DbResult, Producer, RecordKey, RuntimeContext}; use aimdb_knx_connector::dpt::{Dpt1, Dpt9, DptDecode, DptEncode}; -use aimdb_knx_connector::{Channels, KnxConnector}; +use aimdb_knx_connector::KnxConnector; use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use aimdb_uds_connector::UdsServer; @@ -103,17 +103,14 @@ async fn main() -> DbResult<()> { .security_policy(SecurityPolicy::read_only()) .max_connections(10); - // The adapter owns the UDP socket and the clock; the channels are the - // caller's, exactly as on the MCU. - static KNX_CHANNELS: Channels = Channels::new(); - + // The adapter owns the UDP socket and the clock; the connector owns the + // tunnelling protocol. let mut builder = AimDbBuilder::new() .runtime(runtime) .with_connector(KnxConnector::new( TokioNet::udp(Ipv4Addr::UNSPECIFIED), TokioDelay, "knx://192.168.1.4:3671", - &KNX_CHANNELS, )) .with_connector(UdsServer::from_config(remote_config));