diff --git a/Cargo.lock b/Cargo.lock index 1251e840..0c6bae45 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -209,7 +209,6 @@ dependencies = [ "aimdb-embassy-adapter", "aimdb-knx-pico", "aimdb-tokio-adapter", - "async-stream", "critical-section", "defmt 1.1.1", "embassy-executor", @@ -218,13 +217,11 @@ dependencies = [ "embassy-sync", "embassy-time", "futures-core", - "futures-util", "heapless 0.8.0", "static_cell", "thiserror 2.0.17", "tokio", "tokio-test", - "uuid", ] [[package]] diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs index 39e04bc7..8ffabc59 100644 --- a/aimdb-codegen/src/rust.rs +++ b/aimdb-codegen/src/rust.rs @@ -250,7 +250,10 @@ pub fn generate_main_rs(state: &ArchitectureState, binary_name: &str) -> Option< .iter() .filter_map(|c| match c.protocol.as_str() { "mqtt" => Some(quote! { use aimdb_mqtt_connector::MqttConnector; }), - "knx" => Some(quote! { use aimdb_knx_connector::KnxConnector; }), + "knx" => Some(quote! { + use aimdb_knx_connector::{Channels, KnxConnector}; + use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; + }), "ws" => Some(quote! { use aimdb_websocket_connector::WebSocketConnector; }), _ => None, }) @@ -266,7 +269,19 @@ 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) }, - "knx" => quote! { KnxConnector::new(&#var_ident) }, + // The adapter owns the socket and the clock; the channels are + // the binary's, in a block-scoped `static`. + "knx" => quote! { + { + static KNX_CHANNELS: Channels = Channels::new(); + KnxConnector::new( + TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED), + TokioDelay, + &#var_ident, + &KNX_CHANNELS, + ) + } + }, "ws" => quote! { WebSocketConnector::new() .bind(#var_ident.parse::() @@ -474,6 +489,12 @@ pub fn generate_binary_cargo_toml(state: &ArchitectureState, binary_name: &str) let has_knx = bin.external_connectors.iter().any(|c| c.protocol == "knx"); let has_ws = bin.external_connectors.iter().any(|c| c.protocol == "ws"); + let tokio_adapter_features = if has_knx { + "[\"tokio-runtime\", \"net\"]" + } else { + "[\"tokio-runtime\"]" + }; + let mut optional_connector_deps = String::new(); if has_mqtt { optional_connector_deps.push_str( @@ -482,7 +503,9 @@ pub fn generate_binary_cargo_toml(state: &ArchitectureState, binary_name: &str) } if has_knx { optional_connector_deps.push_str( - "aimdb-knx-connector = { version = \"0.5\", features = [\"tokio-runtime\"] }\n", + "# critical-section-std-impl: the KNX channels need an impl, and only \ +the binary may pick one.\n\ +aimdb-knx-connector = { version = \"0.5\", features = [\"tokio-runtime\", \"critical-section-std-impl\"] }\n", ); } if has_ws { @@ -506,7 +529,7 @@ path = \"src/main.rs\"\n\ [dependencies]\n\ {common_crate_dep} = {{ path = \"../{common_crate_name}\" }}\n\ aimdb-core = {{ version = \"0.5\" }}\n\ -aimdb-tokio-adapter = {{ version = \"0.5\", features = [\"tokio-runtime\"] }}\n\ +aimdb-tokio-adapter = {{ version = \"0.5\", features = {tokio_adapter_features} }}\n\ {optional_connector_deps}\ tokio = {{ version = \"1\", features = [\"full\"] }}\n\ tracing = \"0.1\"\n\ @@ -1303,6 +1326,12 @@ pub fn generate_hub_cargo_toml(state: &ArchitectureState) -> String { .iter() .any(|r| r.connectors.iter().any(|c| c.protocol == "ws")); + let tokio_adapter_features = if has_knx { + "[\"tokio-runtime\", \"net\"]" + } else { + "[\"tokio-runtime\"]" + }; + let mut connector_deps = String::new(); if has_mqtt { connector_deps.push_str( @@ -1311,7 +1340,9 @@ pub fn generate_hub_cargo_toml(state: &ArchitectureState) -> String { } if has_knx { connector_deps.push_str( - "aimdb-knx-connector = { version = \"0.5\", features = [\"tokio-runtime\"] }\n", + "# critical-section-std-impl: the KNX channels need an impl, and only \ +the binary may pick one.\n\ +aimdb-knx-connector = { version = \"0.5\", features = [\"tokio-runtime\", \"critical-section-std-impl\"] }\n", ); } if has_ws { @@ -1338,7 +1369,7 @@ path = \"src/main.rs\"\n\ {common_crate_name} = {{ path = \"../{common_crate_name}\" }}\n\ aimdb-core = {{ version = \"0.5\" }}\n\ aimdb-data-contracts = {{ version = \"0.5\", features = [\"linkable\"] }}\n\ -aimdb-tokio-adapter = {{ version = \"0.5\", features = [\"tokio-runtime\"] }}\n\ +aimdb-tokio-adapter = {{ version = \"0.5\", features = {tokio_adapter_features} }}\n\ {connector_deps}\ tokio = {{ version = \"1\", features = [\"full\"] }}\n\ tracing = \"0.1\"\n\ @@ -1379,7 +1410,10 @@ pub fn generate_hub_main_rs(state: &ArchitectureState) -> String { v.push(quote! { use aimdb_mqtt_connector::MqttConnector; }); } if has_knx { - v.push(quote! { use aimdb_knx_connector::KnxConnector; }); + v.push(quote! { + use aimdb_knx_connector::{Channels, KnxConnector}; + use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; + }); } if has_ws { v.push(quote! { use aimdb_websocket_connector::WebSocketConnector; }); @@ -1421,7 +1455,17 @@ pub fn generate_hub_main_rs(state: &ArchitectureState) -> String { v.push(quote! { .with_connector(MqttConnector::new(&mqtt_url)) }); } if has_knx { - v.push(quote! { .with_connector(KnxConnector::new(&knx_gateway)) }); + v.push(quote! { + .with_connector({ + static KNX_CHANNELS: Channels = Channels::new(); + KnxConnector::new( + TokioNet::udp(std::net::Ipv4Addr::UNSPECIFIED), + TokioDelay, + &knx_gateway, + &KNX_CHANNELS, + ) + }) + }); } if has_ws { v.push(quote! { .with_connector(WebSocketConnector::new().bind(ws_bind).path("/ws")) }); diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index cac48efe..9e55df00 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -484,6 +484,7 @@ impl Datagram for EmbassyUdpSocket { } /// Binds [`EmbassyUdpSocket`]s over one caller-owned socket. +#[derive(Clone)] pub struct EmbassyUdpBinder { stack: Stack<'static>, slot: Arc, diff --git a/aimdb-knx-connector/CHANGELOG.md b/aimdb-knx-connector/CHANGELOG.md index 12816867..a228570a 100644 --- a/aimdb-knx-connector/CHANGELOG.md +++ b/aimdb-knx-connector/CHANGELOG.md @@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **One connector for both runtimes (breaking).** + `KnxConnector::new(binder, delay, url, &CHANNELS)` is generic over core's + `DatagramBinder` and `Delay`: a host passes `TokioNet::udp(..)`/`TokioDelay` + where an MCU passes `EmbassyNet::udp(..)`/`EmbassyDelay`. One constructor, no + runtime named in this crate's API, and no `aimdb-tokio-adapter` dependency — + the adapter stays a dev-dependency, as design 052 §8 calls for. + `with_command_queue_size` becomes the const generic `N` — an + `embassy_sync::Channel` is sized at compile time. `tokio_client` and + `embassy_client` are deleted with the `Tokio*`/`Embassy*` aliases, and + `aimdb-codegen` emits the same call with the Tokio transports. - **`tokio-runtime` gains `embassy-sync`; `embassy-futures` is unconditional.** Both are executor-independent, so one channel and select type serves either runtime. diff --git a/aimdb-knx-connector/Cargo.toml b/aimdb-knx-connector/Cargo.toml index 52c7b130..25ae2dd6 100644 --- a/aimdb-knx-connector/Cargo.toml +++ b/aimdb-knx-connector/Cargo.toml @@ -14,14 +14,9 @@ categories = ["network-programming", "embedded", "asynchronous"] [features] default = ["aimdb-core/alloc"] std = ["aimdb-core/std", "knx-pico/std", "thiserror"] -tokio-runtime = [ - "std", - "tokio", - "uuid", - "async-stream", - "futures-util", - "embassy-sync", -] +# The protocol is sans-io and the transports come from an adapter, so the host +# leg adds only `std` and the executor-independent channel type. +tokio-runtime = ["std", "embassy-sync"] # Selects `critical-section`'s std implementation. # @@ -73,19 +68,6 @@ knx-pico = { package = "aimdb-knx-pico", version = "0.3.1", default-features = f # Error handling (std only) thiserror = { workspace = true, optional = true } -# UUID generation for client IDs (std only) -uuid = { version = "1.0", features = ["v4"], optional = true } - -# Tokio runtime dependencies (std) -tokio = { workspace = true, optional = true, features = [ - "sync", - "time", - "net", -] } -async-stream = { version = "0.3", optional = true } -futures-util = { version = "0.3", optional = true, default-features = false, features = [ - "alloc", -] } futures-core = { version = "0.3", default-features = false } # Embassy runtime dependencies (no_std) diff --git a/aimdb-knx-connector/src/connector.rs b/aimdb-knx-connector/src/connector.rs new file mode 100644 index 00000000..cc2aed7a --- /dev/null +++ b/aimdb-knx-connector/src/connector.rs @@ -0,0 +1,314 @@ +//! Runtime-neutral KNX connector. +//! +//! 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. + +use alloc::boxed::Box; +use alloc::string::String; +use alloc::sync::Arc; +use alloc::vec; +use alloc::vec::Vec; +use core::future::Future; +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 embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; +use embassy_sync::channel::Channel; + +use crate::client::{connection_task, shared_channel::ChannelCommands, TelegramSink}; +use crate::tunnel::GroupWrite; + +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 half: the connection task pushes telegrams here for `pump_source`. +struct ChannelTelegrams<'a, const N: usize>(&'a TelegramChannel); + +impl TelegramSink for ChannelTelegrams<'_, N> { + fn try_send(&self, topic: String, payload: Payload) -> bool { + self.0.try_send((topic, payload)).is_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 { + 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 { + /// 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 { + Self { + binder, + delay, + gateway_url: gateway_url.into(), + channels, + } + } + + /// Parse and validate the gateway address. + /// + /// Checked at build so a typo'd IP surfaces as an error rather than a + /// parked connection task. Hostnames are not resolved. + fn gateway_addr(&self) -> DbResult { + let url = ConnectorUrl::parse(&self.gateway_url) + .map_err(|e| DbError::runtime_error(alloc::format!("Invalid KNX URL: {e}")))?; + let port = url.port.unwrap_or(DEFAULT_PORT); + alloc::format!("{}:{}", url.host, port) + .parse() + .map_err(|_| { + DbError::runtime_error(alloc::format!( + "Invalid KNX gateway address {}:{} (an IP address is required; \ + hostnames are not resolved)", + url.host, + port + )) + }) + } +} + +impl ConnectorBuilder for KnxConnector +where + B: aimdb_core::session::DatagramBinder + Clone + Send + Sync + 'static, + D: aimdb_core::session::Delay + Clone + Send + Sync + 'static, +{ + fn build<'a>( + &'a self, + db: &'a AimDb, + ) -> Pin>> + Send + 'a>> { + Box::pin(async move { + let gateway = self.gateway_addr()?; + log_info!("Creating KNX connector for gateway {}", gateway); + + let runtime: Arc = db.runtime_ops(); + let channels = self.channels; + let task: BoxFuture = Box::pin(connection_task( + self.binder.clone(), + gateway, + runtime, + self.delay.clone(), + ChannelTelegrams::(&channels.telegrams), + ChannelCommands::(channels.commands.receiver()), + )); + + let mut futures: Vec = vec![task]; + futures.extend(pump_source( + db, + "knx", + KnxSource:: { + telegrams: &channels.telegrams, + }, + )); + futures.extend(pump_sink( + db, + "knx", + Arc::new(KnxSink:: { + commands: &channels.commands, + }), + )); + Ok(futures) + }) + } + + fn scheme(&self) -> &str { + "knx" + } +} + +#[cfg(all(test, feature = "tokio-runtime"))] +mod tests { + use super::*; + use aimdb_core::buffer::BufferCfg; + use aimdb_core::AimDbBuilder; + use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; + 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| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + builder.build().await.expect("build db").0 + } + + /// A typo'd gateway must fail at `build`, not park a connection task. + #[tokio::test] + async fn an_unparsable_gateway_fails_the_build() { + let db = db().await; + let connector = KnxConnector::<_, _, 8>::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"); + }; + assert!( + format!("{err}").contains("an IP address is required"), + "unexpected error: {err}" + ); + } + + /// The connector registers under the `knx` scheme and contributes the + /// connection task plus its pump futures. + #[tokio::test] + async fn build_yields_the_connection_task_and_pumps() { + 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, + ); + assert_eq!(ConnectorBuilder::scheme(&connector), "knx"); + + let futures = connector.build(&db).await.expect("build"); + assert!( + !futures.is_empty(), + "at least the connection task is contributed" + ); + } + + /// The whole wiring against a real UDP gateway: the task binds, advertises + /// its endpoint, and the handshake reaches the wire. + #[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( + 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(); + + let mut buf = [0u8; 128]; + let (len, _) = tokio::time::timeout( + std::time::Duration::from_secs(5), + gateway.recv_from(&mut buf), + ) + .await + .expect("no CONNECT_REQUEST reached the gateway") + .expect("recv_from"); + + assert!(len >= 14, "CONNECT_REQUEST carries both HPAIs"); + assert_eq!( + u16::from_be_bytes([buf[2], buf[3]]), + 0x0205, + "CONNECT_REQUEST" + ); + assert_ne!( + &buf[8..12], + &[0, 0, 0, 0], + "the real local endpoint is advertised" + ); + + for handle in driving { + handle.abort(); + } + } +} diff --git a/aimdb-knx-connector/src/embassy_client.rs b/aimdb-knx-connector/src/embassy_client.rs deleted file mode 100644 index 7e2b3ddc..00000000 --- a/aimdb-knx-connector/src/embassy_client.rs +++ /dev/null @@ -1,473 +0,0 @@ -//! Embassy transport shim for the KNX/IP connector -//! -//! This module contributes only socket glue for embedded systems: an -//! `embassy-net` UDP socket, the static channels between the pumps and the -//! connection task, and a select loop driving the shared sans-io -//! [`TunnelEngine`]. The entire tunneling -//! lifecycle (handshake, ACK bookkeeping, keepalive, reconnect backoff) lives -//! in [`crate::tunnel`]. -//! -//! # Architecture -//! -//! - **Outbound** (records → telegrams) rides core's `pump_sink` via the -//! [`Connector`](aimdb_core::transport::Connector) impl (commands go onto a -//! `CriticalSectionRawMutex` channel the connection task drains). -//! - **Inbound** (telegrams → records) rides core's `pump_source`: the -//! connection task pushes `(group-address, payload)` onto an inbound channel -//! that `KnxSource` drains. -//! - The connection task is force-`Send`ed once via -//! [`into_box_future`]; the -//! only `unsafe` in this crate is the audited -//! [`NetStack::new`](aimdb_embassy_adapter::connectors::NetStack) call in the -//! builder (single-core cooperative executor invariant). -//! -//! # Usage -//! -//! Illustrative (not compiled: requires the `embassy-runtime` feature and a -//! device network stack): -//! -//! ```rust,ignore -//! use aimdb_knx_connector::KnxConnectorBuilder; -//! use aimdb_core::AimDbBuilder; -//! -//! // Configure database with KNX connector -//! let db = AimDbBuilder::new() -//! .runtime(embassy_adapter) -//! .with_connector( -//! KnxConnectorBuilder::new("knx://192.168.1.19:3671", stack) -//! ) -//! .configure::(|reg| { -//! // Inbound: Monitor KNX bus for light state changes -//! reg.link_from("knx://1/0/7") -//! .with_deserializer(deserialize_light_state) -//! .finish(); -//! }) -//! .build().await?; -//! ``` - -use crate::tunnel::{drain_actions, GroupWrite, TunnelConfig, TunnelEngine, TunnelIo}; -use crate::GroupAddress; -use aimdb_core::connector::ConnectorUrl; -use aimdb_core::session::{pump_sink, pump_source, Payload}; -use aimdb_core::ConnectorBuilder; -use aimdb_embassy_adapter::connectors::into_box_future; -use aimdb_embassy_adapter::SendFutureWrapper; -use alloc::boxed::Box; -use alloc::string::{String, ToString}; -use alloc::sync::Arc; -use alloc::vec::Vec; -use core::future::Future; -use core::pin::Pin; -use core::str::FromStr; -use embassy_net::udp::{PacketMetadata, UdpSocket}; -use embassy_net::{IpAddress, Ipv4Address, Stack}; -use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; -use embassy_sync::channel::{Channel, Receiver, Sender}; -use static_cell::StaticCell; - -/// Inbound telegram item: `(group-address string, payload)` — pushed by the -/// connection task, drained by [`KnxSource`] into core's `pump_source`. -type InboundItem = (String, Payload); - -/// Outbound command item — boxed so the static channel stores pointers -/// instead of full `MAX_APDU`-sized payloads. -type CommandItem = Box; - -/// `'static` reference to the outbound command channel. -type CommandChannelRef = - &'static Channel; -/// Sender / receiver halves of the inbound telegram channel. -type InboundSender = Sender<'static, CriticalSectionRawMutex, InboundItem, KNX_INBOUND_QUEUE_SIZE>; -type InboundReceiver = - Receiver<'static, CriticalSectionRawMutex, InboundItem, KNX_INBOUND_QUEUE_SIZE>; - -/// Capacity of the static KNX command channel. -/// -/// Embassy requires a compile-time const generic — runtime configurability is -/// not possible with `StaticCell>`. Adjust this constant and -/// recompile if your installation needs a larger buffer. -const KNX_COMMAND_QUEUE_SIZE: usize = 32; - -/// Static channel for KNX commands (capacity: [`KNX_COMMAND_QUEUE_SIZE`]) -static KNX_COMMAND_CHANNEL: StaticCell< - Channel, -> = StaticCell::new(); - -/// Get or initialize the command channel -fn get_command_channel( -) -> &'static Channel { - KNX_COMMAND_CHANNEL.init(Channel::new()) -} - -/// Capacity of the static inbound telegram channel. -const KNX_INBOUND_QUEUE_SIZE: usize = 32; - -/// Static channel for inbound telegrams (capacity: [`KNX_INBOUND_QUEUE_SIZE`]). -static KNX_INBOUND_CHANNEL: StaticCell< - Channel, -> = StaticCell::new(); - -/// Get or initialize the inbound telegram channel. -fn get_inbound_channel( -) -> &'static Channel { - KNX_INBOUND_CHANNEL.init(Channel::new()) -} - -/// Inbound [`Source`](aimdb_core::session::Source): drains the connection task's -/// telegram channel, yielding each `(group-address, payload)` for core's -/// `pump_source` to fan out to the matching record producers. The KNX command/ -/// inbound channels use `CriticalSectionRawMutex` (`Send`), so this is a plain -/// `Source` — no force-`Send` wrapper needed. -struct KnxSource { - receiver: InboundReceiver, -} - -impl aimdb_core::session::Source for KnxSource { - fn next(&mut self) -> aimdb_core::session::BoxFut<'_, Option<(String, Payload)>> { - Box::pin(async move { Some(self.receiver.receive().await) }) - } -} - -/// KNX connector builder for Embassy runtime -pub struct KnxConnectorBuilder { - gateway_url: heapless::String<128>, - stack: aimdb_embassy_adapter::connectors::NetStack, -} - -impl KnxConnectorBuilder { - /// Create a new KNX connector builder with gateway URL - /// - /// # Arguments - /// * `gateway_url` - KNX gateway URL (e.g., "knx://192.168.1.19:3671") - /// * `stack` - The device's network stack (the runtime travels as - /// `Arc` and cannot surface it) - pub fn new(gateway_url: &str, stack: &'static Stack<'static>) -> Self { - Self { - gateway_url: heapless::String::try_from(gateway_url) - .unwrap_or_else(|_| heapless::String::new()), - // SAFETY: AimDB's Embassy integration requires a single-core - // cooperative executor (the adapter's module-level invariant); - // every future touching this stack — including the connection - // task built from this builder — is polled on that executor. - stack: unsafe { aimdb_embassy_adapter::connectors::NetStack::new(stack) }, - } - } -} - -type BoxFuture = Pin + Send + 'static>>; - -/// Implement ConnectorBuilder trait for Embassy -impl ConnectorBuilder for KnxConnectorBuilder { - fn build<'a>( - &'a self, - db: &'a aimdb_core::builder::AimDb, - ) -> Pin>> + Send + 'a>> { - // No `.await` here, so the build future is `Send` without a wrapper: the - // tunnelling connection task (which holds the `!Send` UDP socket) is - // force-`Send`ed once via `into_box_future`; the data-flow rides core's - // pumps with `Send` `Connector`/`Source` (KNX channels are - // `CriticalSectionRawMutex`, i.e. `Send`). - Box::pin(async move { - let (command_channel, inbound_rx, connection_task) = - KnxConnectorImpl::setup(self.gateway_url.as_str(), self.stack).map_err(|e| { - #[cfg(feature = "defmt")] - defmt::error!("Failed to build KNX connector"); - aimdb_core::DbError::runtime_error(alloc::format!( - "Failed to build KNX connector: {}", - e - )) - })?; - - // Outbound: records → KNX telegrams via the existing `Connector` impl. - let mut futures = pump_sink(db, "knx", Arc::new(KnxConnectorImpl { command_channel })); - // Inbound: KNX telegrams → records via the connection task's channel. - futures.extend(pump_source( - db, - "knx", - KnxSource { - receiver: inbound_rx, - }, - )); - // The KNX/IP tunnelling state machine (force-`Send` protocol task). - futures.push(connection_task); - - Ok(futures) - }) - } - - fn scheme(&self) -> &str { - "knx" - } -} - -/// Internal KNX connector implementation -pub struct KnxConnectorImpl { - command_channel: CommandChannelRef, -} - -impl KnxConnectorImpl { - /// Set up the command + inbound channels and the tunnelling connection task. - /// - /// Synchronous (no `.await`) so the caller's `build` future stays `Send`. - /// Returns the command channel (for outbound `pump_sink`), the inbound - /// receiver (for `pump_source`), and the force-`Send` connection task. - fn setup( - gateway_url: &str, - stack: aimdb_embassy_adapter::connectors::NetStack, - ) -> Result<(CommandChannelRef, InboundReceiver, BoxFuture), &'static str> { - // Parse the gateway URL - let connector_url = ConnectorUrl::parse(gateway_url).map_err(|_| "Invalid KNX URL")?; - - let host = connector_url.host.clone(); - let port = connector_url.port.unwrap_or(3671); // KNX/IP default port - - #[cfg(feature = "defmt")] - defmt::trace!("Creating KNX connector for {}:{}", host.as_str(), port); - - // Parse gateway IP address - let gateway_ip = Ipv4Address::from_str(&host).map_err(|_| "Invalid gateway IP address")?; - - // Get network stack for background task - let network = stack.get(); - - // Channels: outbound commands (publish → task) and inbound telegrams (task → pump_source). - let command_channel = get_command_channel(); - let inbound_channel = get_inbound_channel(); - let inbound_tx = inbound_channel.sender(); - let inbound_rx = inbound_channel.receiver(); - - // The KNX/IP tunnelling state machine (holds the `!Send` UDP socket — force-`Send`). - let knx_task_future = into_box_future(async move { - #[cfg(feature = "defmt")] - defmt::trace!("KNX background task starting for {}:{}", gateway_ip, port); - - // Run the connection task (this never returns). - connection_task(network, gateway_ip, port, command_channel, inbound_tx).await; - }); - - #[cfg(feature = "defmt")] - defmt::trace!("KNX connector initialized"); - - Ok((command_channel, inbound_rx, knx_task_future)) - } -} - -// Implement the Connector trait -impl aimdb_core::transport::Connector for KnxConnectorImpl { - fn publish( - &self, - resource_id: &str, - _config: &aimdb_core::transport::ConnectorConfig, - payload: &[u8], - ) -> Pin> + Send + '_>> - { - // Validation shared with the tokio shim (same checks, same order); - // boxed so the static channel stores pointers, not full payloads. - let cmd = match GroupWrite::try_new(resource_id, payload) { - Ok(cmd) => Box::new(cmd), - Err(e) => return Box::pin(async move { Err(e) }), - }; - let command_channel = self.command_channel; - - Box::pin(async move { - // Send command to background task via channel - command_channel.send(cmd).await; - - Ok(()) - }) - } -} - -/// Current monotonic time in milliseconds for the engine. -fn now_ms() -> u64 { - embassy_time::Instant::now().as_millis() -} - -/// The connection task: socket I/O around the shared [`TunnelEngine`]. -/// -/// Creates a UDP socket (recreating it whenever the engine asks for a reset), -/// then loops: fire engine deadlines, apply the engine's actions, and select -/// over inbound datagrams, outbound commands, and the next engine deadline. -async fn connection_task( - stack: &'static Stack<'static>, - gateway_addr: Ipv4Address, - gateway_port: u16, - command_channel: CommandChannelRef, - inbound_tx: InboundSender, -) { - let mut engine = TunnelEngine::new(TunnelConfig::default(), now_ms()); - - // Socket buffers outlive each per-connection socket below. - let mut rx_meta = [PacketMetadata::EMPTY; 4]; - let mut rx_buffer = [0; 512]; - let mut tx_meta = [PacketMetadata::EMPTY; 4]; - let mut tx_buffer = [0; 512]; - - loop { - #[cfg(feature = "defmt")] - defmt::info!( - "🔌 Connecting to KNX gateway {}:{}", - gateway_addr, - gateway_port - ); - - let mut socket = UdpSocket::new( - *stack, - &mut rx_meta, - &mut rx_buffer, - &mut tx_meta, - &mut tx_buffer, - ); - - if socket.bind(0).is_err() { - #[cfg(feature = "defmt")] - defmt::error!("Failed to bind KNX socket, retrying in 5s"); - drop(socket); - embassy_time::Timer::after(embassy_time::Duration::from_secs(5)).await; - continue; - } - - // Drive the engine until it asks for a socket reset; the engine is - // then in its backoff phase, so re-entering with a fresh socket only - // reconnects once the backoff deadline passes. - drive_connection( - &mut engine, - &mut socket, - gateway_addr, - gateway_port, - command_channel, - &inbound_tx, - ) - .await; - drop(socket); - - #[cfg(feature = "defmt")] - defmt::trace!("KNX connection reset, reconnecting after backoff..."); - - // Nothing can be sent until the engine's backoff deadline, so wait it - // out before binding the fresh socket. This also paces the rebind - // cycle when a socket errors persistently (the old client likewise - // slept the full backoff between socket teardowns). - let wait_ms = engine.next_deadline().saturating_sub(now_ms()); - embassy_time::Timer::after(embassy_time::Duration::from_millis(wait_ms)).await; - } -} - -/// Socket-side glue for [`drain_actions`]: frames ride the `embassy-net` UDP -/// socket, parsed telegrams ride the static inbound channel into [`KnxSource`]. -struct EmbassyIo<'a, 'b> { - socket: &'a UdpSocket<'b>, - gateway: (IpAddress, u16), - inbound_tx: &'a InboundSender, -} - -impl TunnelIo for EmbassyIo<'_, '_> { - fn send(&mut self, frame: &[u8]) -> impl Future + Send { - // `embassy_net`'s send future is `!Send`; the wrapper is the adapter's - // audited force-`Send`, same single-core invariant as everywhere else. - SendFutureWrapper(async move { - // Log-and-continue: a transient send error must not tear down the - // tunnel; a persistently dead send path surfaces through the - // engine's heartbeat-response timeout. - if self.socket.send_to(frame, self.gateway).await.is_err() { - #[cfg(feature = "defmt")] - defmt::error!("KNX send failed"); - return false; - } - true - }) - } - - fn forward(&mut self, addr: GroupAddress, payload: Vec) { - let resource_id = addr.to_string(); - - #[cfg(feature = "defmt")] - defmt::trace!( - "KNX telegram: {} (len={}) -> routing", - resource_id.as_str(), - payload.len() - ); - - if self - .inbound_tx - .try_send((resource_id, Payload::from(payload))) - .is_err() - { - #[cfg(feature = "defmt")] - defmt::warn!("KNX inbound channel full; dropped telegram"); - } - } - - fn warn_ack_timeout(&mut self, seq: u8) { - let _ = seq; - #[cfg(feature = "defmt")] - defmt::warn!("⚠️ ACK timeout for seq={}", seq); - } -} - -/// Drive the engine over one socket lifetime; returns when the engine asks -/// for a socket reset. -async fn drive_connection( - engine: &mut TunnelEngine, - socket: &mut UdpSocket<'_>, - gateway_addr: Ipv4Address, - gateway_port: u16, - command_channel: CommandChannelRef, - inbound_tx: &InboundSender, -) { - use embassy_futures::select::{select3, Either3}; - - let gateway = (IpAddress::Ipv4(gateway_addr), gateway_port); - - loop { - engine.poll(now_ms()); - - { - let mut io = EmbassyIo { - socket, - gateway, - inbound_tx, - }; - if drain_actions(engine, &mut io).await { - return; - } - } - - let sleep_ms = engine.next_deadline().saturating_sub(now_ms()); - let deadline = embassy_time::Timer::after(embassy_time::Duration::from_millis(sleep_ms)); - let mut recv_buf = [0u8; 512]; - - // Only drain commands while connected: during connect / backoff the - // arm stays pending, so commands keep queueing in the static channel - // and flush once the handshake completes (same as the previous - // implementation, where the select loop only ran while connected). - let connected = engine.is_connected(); - let cmd_arm = async { - if connected { - command_channel.receive().await - } else { - core::future::pending().await - } - }; - - match select3(socket.recv_from(&mut recv_buf), cmd_arm, deadline).await { - Either3::First(Ok((len, _peer))) => { - engine.handle_datagram(&recv_buf[..len], now_ms()); - } - Either3::First(Err(_)) => { - #[cfg(feature = "defmt")] - defmt::error!("Socket receive error"); - engine.handle_socket_error(now_ms()); - } - Either3::Second(cmd) => { - // The command arm only resolves while connected, so the - // engine's disconnected drop path is unreachable here; its - // `false` return is a defensive contract covered by the - // engine unit tests. - let _ = engine.handle_command(*cmd, now_ms()); - } - // Wake for the engine deadline; `poll` at the loop top fires it. - Either3::Third(()) => {} - } - } -} diff --git a/aimdb-knx-connector/src/lib.rs b/aimdb-knx-connector/src/lib.rs index 9ba3b3e3..251a5e9d 100644 --- a/aimdb-knx-connector/src/lib.rs +++ b/aimdb-knx-connector/src/lib.rs @@ -34,8 +34,10 @@ //! ```no_run //! use aimdb_core::buffer::BufferCfg; //! use aimdb_core::AimDbBuilder; -//! use aimdb_knx_connector::KnxConnector; +//! use aimdb_knx_connector::{Channels, KnxConnector}; +//! use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; //! use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +//! use std::net::Ipv4Addr; //! use std::sync::Arc; //! //! #[derive(Debug, Clone)] @@ -46,9 +48,15 @@ //! # 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("knx://192.168.1.19:3671")); +//! .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) //! // Inbound: Monitor KNX bus @@ -78,14 +86,14 @@ //! ```rust,ignore //! use aimdb_core::AimDbBuilder; //! use aimdb_embassy_adapter::EmbassyAdapter; -//! use aimdb_knx_connector::embassy_client::KnxConnectorBuilder; +//! use aimdb_knx_connector::connector::{Channels, KnxConnector}; //! use alloc::sync::Arc; //! //! let runtime = Arc::new(EmbassyAdapter::new()); //! //! let db = AimDbBuilder::new() //! .runtime(runtime) -//! .with_connector(KnxConnectorBuilder::new("knx://192.168.1.19:3671", stack)) +//! .with_connector(KnxConnector::new(binder, EmbassyDelay, gateway, &CHANNELS)) //! .configure::(|reg| { //! reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing) //! .source(sensor_producer) @@ -155,28 +163,9 @@ pub mod tunnel; #[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] pub mod client; -// Platform-specific implementations -#[cfg(feature = "tokio-runtime")] -pub mod tokio_client; - -#[cfg(feature = "embassy-runtime")] -pub mod embassy_client; - -// Re-export platform-specific types -// Both implementations use KnxConnectorBuilder for API consistency -// When both features are enabled (e.g., during testing), prefer tokio -#[cfg(all(feature = "tokio-runtime", not(feature = "embassy-runtime")))] -pub use tokio_client::KnxConnectorBuilder as KnxConnector; - -#[cfg(all(feature = "embassy-runtime", not(feature = "tokio-runtime")))] -pub use embassy_client::KnxConnectorBuilder as KnxConnector; - -// When both features are enabled, export both with different names -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use tokio_client::KnxConnectorBuilder as TokioKnxConnector; - -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use embassy_client::KnxConnectorBuilder as EmbassyKnxConnector; +// Runtime-neutral `KnxConnector` over an adapter's datagram transport. +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +pub mod connector; -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use tokio_client::KnxConnectorBuilder as KnxConnector; // Default to tokio when both enabled +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +pub use connector::{Channels, KnxConnector}; diff --git a/aimdb-knx-connector/src/tokio_client.rs b/aimdb-knx-connector/src/tokio_client.rs deleted file mode 100644 index f68a5dbb..00000000 --- a/aimdb-knx-connector/src/tokio_client.rs +++ /dev/null @@ -1,632 +0,0 @@ -//! Tokio transport shim for the KNX/IP connector -//! -//! This module contributes only socket glue: a UDP socket, the channels -//! between the pumps and the connection task, and a select loop driving the -//! shared sans-io [`TunnelEngine`]. The entire -//! tunneling lifecycle (handshake, ACK bookkeeping, keepalive, reconnect -//! backoff) lives in [`crate::tunnel`]. -//! -//! - Outbound rides core's `pump_sink`: `KnxSink` forwards each serialized -//! record as a [`GroupWrite`] command to the connection task. -//! - Inbound rides core's `pump_source`: the connection task pushes parsed -//! `(group-address, payload)` telegrams that `KnxSource` yields. - -use crate::tunnel::{ - drain_actions, GroupWrite, LocalEndpoint, TunnelConfig, TunnelEngine, TunnelIo, -}; -use crate::GroupAddress; -use aimdb_core::connector::ConnectorUrl; -use aimdb_core::transport::{Connector, ConnectorConfig, PublishError}; -use aimdb_core::{log_debug, log_error, log_info, log_trace, log_warn}; -use aimdb_core::{pump_sink, pump_source, BoxFut, ConnectorBuilder, Payload, Source}; -use std::future::Future; -use std::net::{IpAddr, SocketAddr}; -use std::pin::Pin; -use std::sync::Arc; -use std::time::Duration; -use tokio::net::UdpSocket; -use tokio::sync::mpsc; - -/// KNX connector for a single gateway connection. -/// -/// Each connector manages ONE KNX/IP gateway connection; inbound telegrams are -/// dispatched to AimDB producers by `pump_source`, outbound records published by -/// `pump_sink`. -/// -/// # Usage Pattern -/// -/// The connector collects routes from the database during build() and -/// automatically monitors all required KNX group addresses. -pub struct KnxConnectorBuilder { - gateway_url: String, - /// Capacity of the mpsc channel between outbound publishers and the - /// connection task. Defaults to 32. - command_queue_size: usize, -} - -impl KnxConnectorBuilder { - /// Create a new KNX connector builder - /// - /// # Arguments - /// * `gateway_url` - Gateway URL (knx://host:port) - pub fn new(gateway_url: impl Into) -> Self { - Self { - gateway_url: gateway_url.into(), - command_queue_size: 32, - } - } - - /// Override the internal command channel capacity (default: 32). - /// - /// The channel sits between outbound publisher futures and the single - /// connection task that serializes UDP sends. Increase this for - /// installations with many outbound routes or bursty publish patterns. - pub fn with_command_queue_size(mut self, size: usize) -> Self { - self.command_queue_size = size; - self - } -} - -type BoxFuture = Pin + Send + 'static>>; - -impl ConnectorBuilder for KnxConnectorBuilder { - fn build<'a>( - &'a self, - db: &'a aimdb_core::builder::AimDb, - ) -> Pin>> + Send + 'a>> { - Box::pin(async move { - // Build the command channel, the inbound-telegram channel, and the - // connection task. Inbound flows connection-task → `KnxSource` → - // `pump_source`; outbound flows `pump_sink` → `KnxSink` → the command - // channel → connection task. The routing `Router` is (re)built inside - // `pump_source` from `collect_inbound_routes`. - let (command_tx, telegram_rx, connection_future) = - KnxConnectorImpl::build_internal(&self.gateway_url, self.command_queue_size) - .await - .map_err(|e| { - aimdb_core::DbError::runtime_error(format!( - "Failed to build KNX connector: {}", - e - )) - })?; - - let mut futures: Vec = vec![connection_future]; - // Inbound: the KNX bus source, fanned out to producers by `pump_source`. - futures.extend(pump_source(db, "knx", KnxSource { telegram_rx })); - // Outbound: `pump_sink` serializes each record and hands it to `KnxSink`. - futures.extend(pump_sink(db, "knx", Arc::new(KnxSink { command_tx }))); - - Ok(futures) - }) - } - - fn scheme(&self) -> &str { - "knx" - } -} - -/// Build-time helper aggregating KNX construction logic. -/// -/// `KnxConnectorBuilder::build()` produces a `Vec` containing one -/// connection-task future plus one publisher future per outbound route. -pub struct KnxConnectorImpl; - -impl KnxConnectorImpl { - /// Builds the KNX connection-task future, returning the outbound command - /// sender and the inbound-telegram receiver for `KnxSink` / `KnxSource`. - /// - /// # Arguments - /// * `gateway_url` - Gateway URL (knx://host:port) - /// * `command_queue_size` - Capacity of both the command and telegram channels - async fn build_internal( - gateway_url: &str, - command_queue_size: usize, - ) -> Result< - ( - mpsc::Sender, - mpsc::Receiver<(String, Payload)>, - BoxFuture, - ), - String, - > { - // Parse the gateway URL (bare `knx://host:port`, like the Embassy shim). - let connector_url = - ConnectorUrl::parse(gateway_url).map_err(|e| format!("Invalid KNX URL: {}", e))?; - let gateway_ip = connector_url.host.clone(); - let gateway_port = connector_url.port.unwrap_or(3671); - - // Validate the gateway address here so a typo'd IP (or a hostname — - // never resolved) surfaces as a build() error instead of a parked - // connection task, matching the Embassy shim. - let gateway_addr: SocketAddr = format!("{}:{}", gateway_ip, gateway_port) - .parse() - .map_err(|_| { - format!( - "Invalid KNX gateway address {}:{} (an IP address is required; hostnames are not resolved)", - gateway_ip, gateway_port - ) - })?; - - log_info!("Creating KNX connector for gateway {}", gateway_addr); - - // Outbound commands (publishers → connection task) and inbound telegrams - // (connection task → `KnxSource`/`pump_source`). - let (command_tx, command_rx) = mpsc::channel::(command_queue_size); - let (telegram_tx, telegram_rx) = mpsc::channel::<(String, Payload)>(command_queue_size); - - let connection_future: BoxFuture = - Box::pin(connection_task(gateway_addr, telegram_tx, command_rx)); - - Ok((command_tx, telegram_rx, connection_future)) - } -} - -/// Outbound publish adapter driven by `pump_sink`. -/// -/// `pump_sink` resolves each record's destination group address (dynamic via a -/// topic provider, or the link's default) and serializes the value; `publish` -/// parses that address and forwards a fire-and-forget `GroupValueWrite` to the -/// connection task over the command channel. -struct KnxSink { - command_tx: mpsc::Sender, -} - -impl Connector for KnxSink { - fn publish( - &self, - destination: &str, - _config: &ConnectorConfig, - payload: &[u8], - ) -> Pin> + Send + '_>> { - // Validation shared with the Embassy shim (same checks, same order). - let command = GroupWrite::try_new(destination, payload); - let command_tx = self.command_tx.clone(); - Box::pin(async move { - command_tx - .send(command?) - .await - .map_err(|_| PublishError::ConnectionFailed) // connection task gone - }) - } -} - -/// Inbound telegram source driven by `pump_source`. -/// -/// Yields each `(group_address, payload)` the connection task parsed off the KNX -/// bus; `pump_source` deserializes and fans it out to the matching producers. -struct KnxSource { - telegram_rx: mpsc::Receiver<(String, Payload)>, -} - -impl Source for KnxSource { - fn next(&mut self) -> BoxFut<'_, Option<(String, Payload)>> { - Box::pin(async move { self.telegram_rx.recv().await }) - } -} - -/// The connection task: socket I/O around the shared [`TunnelEngine`]. -/// -/// Each outer iteration binds a fresh UDP socket and drives the engine over -/// its lifetime: fire engine deadlines, apply the engine's actions, and select -/// over inbound datagrams, outbound commands, and the next engine deadline. -/// When the engine asks for a socket reset, the socket is dropped and the -/// engine's backoff deadline is waited out before rebinding. -async fn connection_task( - gateway_addr: SocketAddr, - telegram_tx: mpsc::Sender<(String, Payload)>, - mut command_rx: mpsc::Receiver, -) { - log_info!("KNX connection task started for {}", gateway_addr); - - let epoch = tokio::time::Instant::now(); - let now_ms = || epoch.elapsed().as_millis() as u64; - - let mut engine = TunnelEngine::new(TunnelConfig::default(), now_ms()); - let mut buf = [0u8; 1024]; - // Set to false once every `KnxSink` is gone. With no outbound routes that - // happens right at build time (`pump_sink` drops the unused sink), so a - // closed command channel only disables its select arm — inbound routing - // keeps running. - let mut commands_open = true; - - loop { - // Bind a fresh socket for this connection cycle and advertise its - // real address in the next CONNECT_REQUEST. - let socket = match UdpSocket::bind("0.0.0.0:0").await { - Ok(s) => { - if let Ok(local) = s.local_addr() { - if let IpAddr::V4(ip) = local.ip() { - engine.set_local_endpoint(LocalEndpoint::Explicit { - ip: ip.octets(), - port: local.port(), - }); - } - log_debug!("KNX: Connecting from {} to {}", local, gateway_addr); - } - s - } - Err(_e) => { - log_error!("Failed to bind UDP socket: {}, retrying in 5s", _e); - tokio::time::sleep(Duration::from_secs(5)).await; - continue; - } - }; - - // Drive the engine over this socket's lifetime. - loop { - engine.poll(now_ms()); - - let mut io = TokioIo { - socket: &socket, - gateway: gateway_addr, - telegram_tx: &telegram_tx, - }; - if drain_actions(&mut engine, &mut io).await { - break; // engine asked for a socket reset - } - - let sleep_ms = engine.next_deadline().saturating_sub(now_ms()); - - tokio::select! { - result = socket.recv_from(&mut buf) => match result { - Ok((len, _)) => { - log_trace!("Received {} bytes from gateway", len); - engine.handle_datagram(&buf[..len], now_ms()); - } - Err(_e) => { - log_error!("Socket error: {}", _e); - engine.handle_socket_error(now_ms()); - } - }, - // Only drained while connected: commands queue up in the channel - // during a reconnect cycle and flush once the handshake completes - // (same as the previous implementation, where the select loop only - // ran while connected). - cmd = command_rx.recv(), if commands_open && engine.is_connected() => match cmd { - Some(cmd) => { - // The arm guard above only admits commands while - // connected, so the engine's disconnected drop path is - // unreachable here; its `false` return is a defensive - // contract covered by the engine unit tests. - let _ = engine.handle_command(cmd, now_ms()); - } - // All `KnxSink`s dropped — no outbound publisher remains. - // Inbound monitoring still has to run, so only disable this - // arm instead of exiting the connection task. - None => commands_open = false, - }, - // Wake for the next engine deadline; `poll` at the loop top fires it. - _ = tokio::time::sleep(Duration::from_millis(sleep_ms)) => {} - } - } - - log_error!("KNX connection lost, reconnecting after backoff..."); - // The engine is backing off: nothing can be sent until its deadline, - // so wait it out before binding the fresh socket. This also paces the - // rebind cycle when a socket errors persistently (the old client - // likewise slept the full backoff between socket teardowns). - let wait_ms = engine.next_deadline().saturating_sub(now_ms()); - tokio::time::sleep(Duration::from_millis(wait_ms)).await; - } -} - -/// Socket-side glue for [`drain_actions`]: frames ride the bound UDP socket, -/// parsed telegrams ride the mpsc channel into [`KnxSource`]. -struct TokioIo<'a> { - socket: &'a UdpSocket, - gateway: SocketAddr, - telegram_tx: &'a mpsc::Sender<(String, Payload)>, -} - -impl TunnelIo for TokioIo<'_> { - async fn send(&mut self, frame: &[u8]) -> bool { - // Log-and-continue: a transient send error must not tear down the - // tunnel; a persistently dead send path surfaces through the engine's - // heartbeat-response timeout. - match self.socket.send_to(frame, self.gateway).await { - Ok(_) => true, - Err(_e) => { - log_error!("KNX send failed: {}", _e); - false - } - } - } - - fn forward(&mut self, addr: GroupAddress, payload: Vec) { - log_debug!("KNX telegram: {} ({} bytes)", addr, payload.len()); - - if self - .telegram_tx - .try_send((addr.to_string(), Payload::from(payload))) - .is_err() - { - log_warn!( - "KNX inbound: dropping telegram for {} (channel full/closed)", - addr - ); - } - } - - fn warn_ack_timeout(&mut self, _seq: u8) { - log_warn!("⚠️ ACK timeout for seq={}", _seq); - } -} - -#[cfg(test)] -mod tests { - use super::*; - use tokio::time::timeout; - - const RECV_TIMEOUT: Duration = Duration::from_secs(5); - - fn service_type_of(frame: &[u8]) -> u16 { - u16::from_be_bytes([frame[2], frame[3]]) - } - - /// CONNECT_RESPONSE: header + [channel_id, status, HPAI(8), CRD(4)]. - fn connect_response(channel_id: u8, status: u8) -> Vec { - let mut frame = vec![0x06, 0x10, 0x02, 0x06, 0x00, 0x14]; - frame.extend_from_slice(&[channel_id, status]); - frame.extend_from_slice(&[0x08, 0x01, 0, 0, 0, 0, 0, 0]); // HPAI 0.0.0.0:0 - frame.extend_from_slice(&[0x04, 0x04, 0x02, 0x00]); // CRD: tunnel - frame - } - - /// TUNNELING_REQUEST carrying a 6-bit GroupValueWrite to 1/0/7 (value 1). - fn inbound_group_write(channel_id: u8, seq: u8) -> Vec { - let cemi = [ - 0x29, 0x00, 0xBC, 0xE0, // L_Data.ind, no add-info, ctrl1, ctrl2 - 0x00, 0x00, 0x08, 0x07, // src 0.0.0, dest 1/0/7 - 0x01, 0x00, 0x81, // NPDU len, TPCI, APCI | value 1 - ]; - let total = 6 + 4 + cemi.len() as u16; - let mut frame = vec![0x06, 0x10, 0x04, 0x20]; - frame.extend_from_slice(&total.to_be_bytes()); - frame.extend_from_slice(&[0x04, channel_id, seq, 0x00]); // connection header - frame.extend_from_slice(&cemi); - frame - } - - /// TUNNELING_ACK from the gateway: header + connection header. - fn gateway_ack(channel_id: u8, seq: u8) -> Vec { - vec![ - 0x06, 0x10, 0x04, 0x21, 0x00, 0x0A, // header, total len 10 - 0x04, channel_id, seq, 0x00, // connection header, status OK - ] - } - - /// Scenario: the gateway drops the first ACK; the client retransmits the - /// byte-identical TUNNELING_REQUEST (same sequence counter, KNXnet/IP - /// 3.8.4) after the ACK timeout, and the tunnel survives once the repeat - /// is ACKed. Real-time test: waits out the 3 s default ACK timeout. - #[tokio::test] - async fn dropped_ack_triggers_identical_retransmit() { - let gateway = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let gateway_port = gateway.local_addr().unwrap().port(); - - let (command_tx, mut telegram_rx, connection_future) = - KnxConnectorImpl::build_internal(&format!("knx://127.0.0.1:{}", gateway_port), 8) - .await - .unwrap(); - let task = tokio::spawn(connection_future); - - let mut buf = [0u8; 1024]; - let (len, client_addr) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no CONNECT_REQUEST") - .unwrap(); - assert_eq!(service_type_of(&buf[..len]), 0x0205); - gateway - .send_to(&connect_response(7, 0), client_addr) - .await - .unwrap(); - - // Outbound write; deliberately do NOT ACK the first request. - let mut data = heapless::Vec::new(); - data.push(0x01).unwrap(); - command_tx - .send(GroupWrite { - group_addr: "1/0/8".parse().unwrap(), - data, - }) - .await - .unwrap(); - let (len, _) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no TUNNELING_REQUEST") - .unwrap(); - let first = buf[..len].to_vec(); - assert_eq!(service_type_of(&first), 0x0420); - - // The retransmit arrives after the ACK timeout, byte-identical. - let (len, _) = timeout(Duration::from_secs(8), gateway.recv_from(&mut buf)) - .await - .expect("no retransmit after dropped ACK") - .unwrap(); - assert_eq!(&buf[..len], &first[..]); - - // ACK the repeat: the tunnel stays up — an inbound telegram still - // round-trips on the same channel (a disconnect would have produced - // a CONNECT_REQUEST here instead of an ACK). - gateway - .send_to(&gateway_ack(7, 0), client_addr) - .await - .unwrap(); - gateway - .send_to(&inbound_group_write(7, 42), client_addr) - .await - .unwrap(); - let (len, _) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no TUNNELING_ACK for inbound telegram") - .unwrap(); - assert_eq!(service_type_of(&buf[..len]), 0x0421); - let (topic, _) = timeout(RECV_TIMEOUT, telegram_rx.recv()) - .await - .expect("no telegram routed") - .unwrap(); - assert_eq!(topic, "1/0/7"); - - task.abort(); - } - - /// Full roundtrip against a scripted fake gateway on localhost UDP: - /// handshake, inbound telegram → `KnxSource` channel, outbound command → - /// TUNNELING_REQUEST on the wire (then ACKed). - #[tokio::test] - async fn tunnel_roundtrip_against_fake_gateway() { - let gateway = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let gateway_port = gateway.local_addr().unwrap().port(); - - let (command_tx, mut telegram_rx, connection_future) = - KnxConnectorImpl::build_internal(&format!("knx://127.0.0.1:{}", gateway_port), 8) - .await - .unwrap(); - let task = tokio::spawn(connection_future); - - // Handshake: CONNECT_REQUEST in, CONNECT_RESPONSE out. - let mut buf = [0u8; 1024]; - let (len, client_addr) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no CONNECT_REQUEST") - .unwrap(); - assert_eq!(service_type_of(&buf[..len]), 0x0205); - gateway - .send_to(&connect_response(7, 0), client_addr) - .await - .unwrap(); - - // Inbound: gateway pushes a telegram; the client ACKs it and the - // parsed payload reaches the telegram channel. - gateway - .send_to(&inbound_group_write(7, 42), client_addr) - .await - .unwrap(); - let (len, _) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no TUNNELING_ACK") - .unwrap(); - assert_eq!(service_type_of(&buf[..len]), 0x0421); - assert_eq!(buf[8], 42); // sequence echoed - let (topic, payload) = timeout(RECV_TIMEOUT, telegram_rx.recv()) - .await - .expect("no telegram routed") - .unwrap(); - assert_eq!(topic, "1/0/7"); - assert_eq!(&payload[..], &[0x01]); - - // Outbound: a GroupWrite command becomes a TUNNELING_REQUEST. - let mut data = heapless::Vec::new(); - data.push(0x01).unwrap(); - command_tx - .send(GroupWrite { - group_addr: "1/0/8".parse().unwrap(), - data, - }) - .await - .unwrap(); - let (len, _) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no TUNNELING_REQUEST") - .unwrap(); - assert_eq!(service_type_of(&buf[..len]), 0x0420); - assert_eq!(buf[8], 0); // first outbound sequence - assert_eq!(&buf[16..18], &[0x08, 0x08]); // cEMI destination = 1/0/8 - assert_eq!(buf[len - 1], 0x81); // APCI GroupValueWrite | 6-bit value 1 - - task.abort(); - } - - /// Inbound-only regression: with no outbound routes, `pump_sink` drops the - /// only `KnxSink` (and with it the sole command sender) at build time. The - /// connection task must keep routing inbound telegrams — a closed command - /// channel only disables that select arm. - #[tokio::test] - async fn inbound_routing_survives_dropped_command_sender() { - let gateway = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let gateway_port = gateway.local_addr().unwrap().port(); - - let (command_tx, mut telegram_rx, connection_future) = - KnxConnectorImpl::build_internal(&format!("knx://127.0.0.1:{}", gateway_port), 8) - .await - .unwrap(); - drop(command_tx); // inbound-only configuration - let task = tokio::spawn(connection_future); - - let mut buf = [0u8; 1024]; - let (len, client_addr) = timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) - .await - .expect("no CONNECT_REQUEST") - .unwrap(); - assert_eq!(service_type_of(&buf[..len]), 0x0205); - gateway - .send_to(&connect_response(7, 0), client_addr) - .await - .unwrap(); - - // The telegram arrives after the handshake completed — the moment the - // old code observed the closed channel and exited. - gateway - .send_to(&inbound_group_write(7, 1), client_addr) - .await - .unwrap(); - let (topic, payload) = timeout(RECV_TIMEOUT, telegram_rx.recv()) - .await - .expect("connection task died — no telegram routed") - .unwrap(); - assert_eq!(topic, "1/0/7"); - assert_eq!(&payload[..], &[0x01]); - - task.abort(); - } - - #[tokio::test] - async fn test_connector_creation() { - let connector = KnxConnectorImpl::build_internal("knx://192.168.1.19:3671", 32).await; - assert!(connector.is_ok()); - } - - #[tokio::test] - async fn test_connector_rejects_hostname_at_build() { - // Hostnames are never resolved (`SocketAddr::parse` only accepts IP - // addresses), so this must fail from build() instead of producing a - // connector whose task can never reach a gateway. - let connector = KnxConnectorImpl::build_internal("knx://gateway.local:3672", 32).await; - assert!(connector.is_err()); - } - - #[test] - fn test_group_address_parsing() { - // Test using knx-pico's GroupAddress parser - assert_eq!("1/0/7".parse::().unwrap().raw(), 0x0807); - assert_eq!("0/0/0".parse::().unwrap().raw(), 0x0000); - assert_eq!("31/7/255".parse::().unwrap().raw(), 0xFFFF); - - // knx-pico supports both 3-level (main/middle/sub) and 2-level (main/sub) formats - assert!("1/0".parse::().is_ok()); // 2-level format is valid - - // Invalid formats - assert!("32/0/0".parse::().is_err()); // main > 31 - assert!("0/8/0".parse::().is_err()); // middle > 7 in 3-level - assert!("invalid".parse::().is_err()); // not a number - } - - #[test] - fn test_group_address_formatting() { - // Test using knx-pico's GroupAddress Display impl - assert_eq!(GroupAddress::from(0x0807).to_string(), "1/0/7"); - assert_eq!(GroupAddress::from(0x0000).to_string(), "0/0/0"); - assert_eq!(GroupAddress::from(0xFFFF).to_string(), "31/7/255"); - } - - #[test] - fn test_group_address_roundtrip() { - let addresses = vec!["1/0/7", "0/0/0", "31/7/255", "5/3/128"]; - - for addr in addresses { - let parsed = addr.parse::().unwrap(); - let formatted = parsed.to_string(); - assert_eq!(formatted, addr); - } - } -} diff --git a/aimdb-knx-connector/tests/topic_provider_tests.rs b/aimdb-knx-connector/tests/topic_provider_tests.rs index f5f2f8c1..b3c8e6d4 100644 --- a/aimdb-knx-connector/tests/topic_provider_tests.rs +++ b/aimdb-knx-connector/tests/topic_provider_tests.rs @@ -11,10 +11,17 @@ 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 // ============================================================================ @@ -317,7 +324,12 @@ 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("knx://192.168.1.10:3671"), + aimdb_knx_connector::KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, + "knx://192.168.1.10:3671", + &CHANNELS, + ), ); // Register dimmer with dynamic group address provider @@ -346,7 +358,12 @@ async fn test_knx_topic_resolver_with_connector_registration() { std::env::set_var("KNX_SWITCH_INPUT", "1/2/10"); let mut builder = AimDbBuilder::new().runtime(runtime).with_connector( - aimdb_knx_connector::KnxConnector::new("knx://192.168.1.10:3671"), + aimdb_knx_connector::KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, + "knx://192.168.1.10:3671", + &CHANNELS, + ), ); // Register switch with dynamic group address resolver @@ -373,7 +390,12 @@ 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("knx://192.168.1.10:3671"), + aimdb_knx_connector::KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, + "knx://192.168.1.10:3671", + &CHANNELS, + ), ); // HVAC setpoint with zone-based routing diff --git a/aimdb-tokio-adapter/src/net.rs b/aimdb-tokio-adapter/src/net.rs index 574d6f09..f24a4e4d 100644 --- a/aimdb-tokio-adapter/src/net.rs +++ b/aimdb-tokio-adapter/src/net.rs @@ -146,6 +146,7 @@ impl Datagram for TokioDatagram { } /// Binds [`TokioDatagram`]s, one per reconnect cycle. +#[derive(Clone, Copy)] pub struct TokioUdpBinder { local_ip: IpAddr, } diff --git a/examples/embassy-knx-connector-demo/src/main.rs b/examples/embassy-knx-connector-demo/src/main.rs index 84e4c032..7b60db2c 100644 --- a/examples/embassy-knx-connector-demo/src/main.rs +++ b/examples/embassy-knx-connector-demo/src/main.rs @@ -42,13 +42,15 @@ extern crate alloc; use aimdb_core::remote::SecurityPolicy; use aimdb_core::{AimDbBuilder, RecordKey, RuntimeContext}; +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::dpt::{Dpt1, Dpt9, DptDecode, DptEncode}; -use aimdb_knx_connector::embassy_client::KnxConnectorBuilder; use aimdb_serial_connector::embassy_transport::SerialServer; use defmt::*; use embassy_executor::Spawner; use embassy_net::StackResources; +use embassy_net::udp::PacketMetadata; use embassy_stm32::eth::{Ethernet, GenericPhy, PacketQueue}; use embassy_stm32::exti::{self, ExtiInput}; use embassy_stm32::gpio::{Level, Output, Pull, Speed}; @@ -269,11 +271,31 @@ async fn main(spawner: Spawner) { .unwrap(); 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. + 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]), + KNX_RX_BUF.init([0; 1024]), + KNX_TX_META.init([PacketMetadata::EMPTY; 8]), + KNX_TX_BUF.init([0; 1024]), + ); + // Read-only: KNX owns the writer for every record (single-writer-per-key), so // remote `record.set` is refused — peers can list/get/subscribe, not write. let mut builder = AimDbBuilder::new() .runtime(runtime.clone()) - .with_connector(KnxConnectorBuilder::new(&gateway_url, stack)) + .with_connector(KnxConnector::new( + knx_binder, + EmbassyDelay, + &gateway_url, + &KNX_CHANNELS, + )) .with_connector( SerialServer::new(serial_rx, serial_tx).security_policy(SecurityPolicy::read_only()), ); diff --git a/examples/tokio-knx-connector-demo/Cargo.toml b/examples/tokio-knx-connector-demo/Cargo.toml index 87d5cd6b..ebb2c44f 100644 --- a/examples/tokio-knx-connector-demo/Cargo.toml +++ b/examples/tokio-knx-connector-demo/Cargo.toml @@ -16,6 +16,8 @@ tracing = ["dep:tracing", "dep:tracing-subscriber"] aimdb-core = { path = "../../aimdb-core", features = ["std", "derive"] } aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = [ "tokio-runtime", + # `TokioNet`/`TokioDelay`: the adapter owns the socket and the clock. + "net", "tracing", ] } @@ -28,6 +30,9 @@ knx-connector-demo-common = { path = "../knx-connector-demo-common", features = # KNX connector aimdb-knx-connector = { path = "../../aimdb-knx-connector", features = [ "tokio-runtime", + # The connector's channels are `CriticalSectionRawMutex`; only the final + # binary may pick the impl they need to link. + "critical-section-std-impl", "tracing", ] } diff --git a/examples/tokio-knx-connector-demo/src/main.rs b/examples/tokio-knx-connector-demo/src/main.rs index c9e5ebab..fe266d9b 100644 --- a/examples/tokio-knx-connector-demo/src/main.rs +++ b/examples/tokio-knx-connector-demo/src/main.rs @@ -25,8 +25,11 @@ 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_tokio_adapter::net::{TokioDelay, TokioNet}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use aimdb_uds_connector::UdsServer; +use std::net::Ipv4Addr; use std::sync::Arc; use tokio::io::{AsyncBufReadExt, BufReader}; @@ -100,10 +103,17 @@ 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(); + let mut builder = AimDbBuilder::new() .runtime(runtime) - .with_connector(aimdb_knx_connector::KnxConnector::new( + .with_connector(KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, "knx://192.168.1.4:3671", + &KNX_CHANNELS, )) .with_connector(UdsServer::from_config(remote_config));