diff --git a/CHANGELOG.md b/CHANGELOG.md index 27905f85..e44f0ae8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`ConnectorBuilder::owns_scheme` — a connector may declare that it must be the + only one registered under its scheme.** Routes are collected by scheme alone + (`collect_inbound_routes`, `collect_outbound_routes`, `pump_source`, + `pump_sink`, `pump_client`), so two connectors that each collect their + scheme's routes both claim *all* of them: every `link_to` gets two publishers, + and nothing in a route says which connector it belongs to. `AimDbBuilder::build` + now rejects that before building any connector, with an error naming the + scheme. Defaults to `false`, so nothing changes for a connector that collects + no routes — a session *server* binds its own listener, and two of them under + one scheme remain two endpoints onto one dispatch. `aimdb-knx-connector` opts + in; MQTT and the session clients are candidates but are left alone for now. - **`AimDbHandle::shutdown(&self)` / `is_closed()`.** The shutdown contract a foreign-language binding needs, moved into the crate whose thread it is about; pinned by `aimdb-sync/tests/shutdown_contract_test.rs`. `detach(self)` diff --git a/Cargo.lock b/Cargo.lock index c4dec9cf..6070576a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -206,25 +206,14 @@ name = "aimdb-knx-connector" version = "0.5.0" dependencies = [ "aimdb-core", - "aimdb-embassy-adapter", "aimdb-knx-pico", "aimdb-tokio-adapter", - "async-stream", "critical-section", - "defmt 1.1.1", - "embassy-executor", "embassy-futures 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", - "embassy-net", "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/Makefile b/Makefile index 462ebf7a..2001c59c 100644 --- a/Makefile +++ b/Makefile @@ -119,7 +119,7 @@ build: @printf "$(YELLOW) → Building persistence SQLite backend$(NC)\n" cargo build --package aimdb-persistence-sqlite @printf "$(YELLOW) → Building KNX connector$(NC)\n" - cargo build --package aimdb-knx-connector --features "std,tokio-runtime" + cargo build --package aimdb-knx-connector --no-default-features --features "std" @printf "$(YELLOW) → Building WebSocket connector (server + client)$(NC)\n" cargo build --package aimdb-websocket-connector --features "server,client" @printf "$(YELLOW) → Building UDS connector$(NC)\n" @@ -212,7 +212,7 @@ test: @printf "$(YELLOW) → Testing MQTT connector (tokio + rustls)$(NC)\n" cargo test --package aimdb-mqtt-connector --features "std,tokio-runtime,tokio-rustls" @printf "$(YELLOW) → Testing KNX connector$(NC)\n" - cargo test --package aimdb-knx-connector --features "std,tokio-runtime" + cargo test --package aimdb-knx-connector --no-default-features --features "std" @printf "$(YELLOW) → Testing WebSocket connector (server + client: unit, real-socket e2e, AimDB round-trip)$(NC)\n" cargo test --package aimdb-websocket-connector --features "server,client" @printf "$(YELLOW) → Testing WebSocket connector client-only build$(NC)\n" @@ -323,9 +323,9 @@ clippy: @printf "$(YELLOW) → Clippy on persistence SQLite backend$(NC)\n" cargo clippy --package aimdb-persistence-sqlite --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on KNX connector (std)$(NC)\n" - cargo clippy --package aimdb-knx-connector --features "std,tokio-runtime" --all-targets -- -D warnings - @printf "$(YELLOW) → Clippy on KNX connector (embassy)$(NC)\n" - cargo clippy --package aimdb-knx-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime" -- -D warnings + cargo clippy --package aimdb-knx-connector --no-default-features --features "std" --all-targets -- -D warnings + @printf "$(YELLOW) → Clippy on KNX connector (neutral, no_std+alloc)$(NC)\n" + cargo clippy --package aimdb-knx-connector --target thumbv7em-none-eabihf --no-default-features --features "connector" -- -D warnings @printf "$(YELLOW) → Clippy on MQTT connector (tokio, no TLS backend)$(NC)\n" cargo clippy --package aimdb-mqtt-connector --features "std,tokio-runtime" --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on MQTT connector (tokio + native-tls)$(NC)\n" @@ -336,8 +336,8 @@ clippy: cargo clippy --package aimdb-mqtt-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime,defmt" -- -D warnings @printf "$(YELLOW) → Clippy on MQTT connector (embassy + TLS + defmt)$(NC)\n" cargo clippy --package aimdb-mqtt-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime,embassy-tls,defmt" -- -D warnings - @printf "$(YELLOW) → Clippy on KNX connector (embassy + defmt)$(NC)\n" - cargo clippy --package aimdb-knx-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime,defmt" -- -D warnings + @printf "$(YELLOW) → Clippy on KNX connector (neutral + defmt)$(NC)\n" + cargo clippy --package aimdb-knx-connector --target thumbv7em-none-eabihf --no-default-features --features "connector,defmt" -- -D warnings @printf "$(YELLOW) → Clippy on WebSocket connector$(NC)\n" cargo clippy --package aimdb-websocket-connector --features "tokio-runtime,client" --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on UDS connector$(NC)\n" @@ -377,7 +377,7 @@ doc: cargo doc --package aimdb-tokio-adapter --features "tokio-runtime,tracing,observability,net" --no-deps cargo doc --package aimdb-sync --no-deps cargo doc --package aimdb-mqtt-connector --features "std,tokio-runtime" --no-deps - cargo doc --package aimdb-knx-connector --features "std,tokio-runtime" --no-deps + cargo doc --package aimdb-knx-connector --no-default-features --features "std" --no-deps cargo doc --package aimdb-codegen --no-deps cargo doc --package aimdb-cli --no-deps cargo doc --package aimdb-mcp --no-deps @@ -398,7 +398,7 @@ doc: cargo doc --package aimdb-core --no-default-features --features alloc --no-deps cargo doc --package aimdb-embassy-adapter --features "embassy-runtime,net" --no-deps cargo doc --package aimdb-mqtt-connector --no-default-features --features "embassy-runtime" --no-deps - cargo doc --package aimdb-knx-connector --no-default-features --features "embassy-runtime" --no-deps + cargo doc --package aimdb-knx-connector --no-default-features --features "connector" --no-deps cargo doc --package aimdb-serial-connector --no-default-features --features "connector" --no-deps cargo doc --package aimdb-tcp-connector --no-default-features --features "connector" --no-deps @cp -r target/doc/* target/doc-final/embedded/ @@ -464,10 +464,10 @@ test-embedded: cargo check --package aimdb-mqtt-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime" @printf "$(YELLOW) → Checking aimdb-mqtt-connector (Embassy + defmt) on thumbv7em-none-eabihf target$(NC)\n" cargo check --package aimdb-mqtt-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime,defmt" - @printf "$(YELLOW) → Checking aimdb-knx-connector (Embassy) on thumbv7em-none-eabihf target$(NC)\n" - cargo check --package aimdb-knx-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime" - @printf "$(YELLOW) → Checking aimdb-knx-connector (Embassy + defmt) on thumbv7em-none-eabihf target$(NC)\n" - cargo check --package aimdb-knx-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime,defmt" + @printf "$(YELLOW) → Checking aimdb-knx-connector (neutral, no_std+alloc) on thumbv7em-none-eabihf target$(NC)\n" + cargo check --package aimdb-knx-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "connector" + @printf "$(YELLOW) → Checking aimdb-knx-connector (neutral + defmt) on thumbv7em-none-eabihf target$(NC)\n" + cargo check --package aimdb-knx-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "connector,defmt" @printf "$(YELLOW) → Checking aimdb-serial-connector (Embassy: full no_std AimX serial client+server) on thumbv7em-none-eabihf target$(NC)\n" cargo check --package aimdb-serial-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "_test-embassy" @printf "$(YELLOW) → Checking aimdb-serial-connector (Embassy + defmt) on thumbv7em-none-eabihf target$(NC)\n" diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs index 39e04bc7..bd2c42b7 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 = [\"std\", \"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 = [\"std\", \"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-core/src/builder.rs b/aimdb-core/src/builder.rs index 3500c443..635e8773 100644 --- a/aimdb-core/src/builder.rs +++ b/aimdb-core/src/builder.rs @@ -774,6 +774,45 @@ impl AimDbBuilder { // transport, applies the security policy's writable marking, and drives // the shared session engine. See `with_connector`'s docs. + // A connector that claims every route for its scheme must be the only + // one registered under it (see `ConnectorBuilder::owns_scheme`). + // Checked before any `build`, so the error names the misconfiguration + // instead of surfacing later as duplicated or misdirected traffic. + // Keyless `ConfigError`s: the mistake is the db's, not a record's. + let mut duplicate_schemes: Vec = Vec::new(); + for (i, builder) in self.connector_builders.iter().enumerate() { + if !builder.owns_scheme() { + continue; + } + let first = self.connector_builders[..i] + .iter() + .position(|earlier| earlier.scheme() == builder.scheme()); + // Only the second registration reports; a third would repeat it. + if first.is_some() + && !duplicate_schemes + .iter() + .any(|e| e.url.as_deref() == Some(builder.scheme())) + { + duplicate_schemes.push(crate::error::ConfigError::new( + "", + Some(builder.scheme().into()), + alloc::format!( + "More than one connector registered for scheme '{}'. Routes are \ + collected by scheme alone, so each connector would claim all of \ + them: every `link_to` would publish twice, and the routes cannot \ + be divided between the two. Register one connector for this \ + scheme, or give one of them a distinct scheme.", + builder.scheme() + ), + )); + } + } + if !duplicate_schemes.is_empty() { + return Err(DbError::InvalidConfiguration { + errors: duplicate_schemes, + }); + } + // Collect connector futures. Connector builders return a // `Vec` for the runner to drive — there is no connector // object to keep. diff --git a/aimdb-core/src/connector.rs b/aimdb-core/src/connector.rs index f54e37ab..fa4c521a 100644 --- a/aimdb-core/src/connector.rs +++ b/aimdb-core/src/connector.rs @@ -779,6 +779,28 @@ pub trait ConnectorBuilder: Send + Sync { /// will be registered under. Used for routing `.link_from()` and `.link_to()` /// declarations to the appropriate connector. fn scheme(&self) -> &str; + + /// Whether registering a second connector under this scheme is an error. + /// + /// Say `true` when [`build`](Self::build) claims every route for its + /// scheme — [`collect_inbound_routes`](crate::AimDb::collect_inbound_routes), + /// [`collect_outbound_routes`](crate::AimDb::collect_outbound_routes), + /// and `crate::session`'s `pump_source`, `pump_sink` and `pump_client` + /// (left unlinked: that module is behind `connector-session`, and this + /// trait is not) all filter by scheme alone, so two such connectors each + /// collect *all* of it: every `link_to` gets two publishers, and the routes + /// cannot be divided between the two endpoints because nothing in a route + /// names which connector it belongs to. That misconfiguration is otherwise + /// silent, and it fails as duplicated or misdirected traffic at runtime + /// rather than at build. + /// + /// Leave it `false` — the default — for a connector that only serves what + /// it is given, such as a session *server*: it binds its own listener and + /// collects no routes, so two of them under one scheme are two endpoints + /// onto the same dispatch, which is useful rather than broken. + fn owns_scheme(&self) -> bool { + false + } } #[cfg(test)] diff --git a/aimdb-core/src/typed_api.rs b/aimdb-core/src/typed_api.rs index 22e6f778..99a40476 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -1596,6 +1596,76 @@ mod tests { } } + /// A connector that claims every route for its scheme, as KNX and the + /// session *clients* do. + struct OwningConnectorBuilder(&'static str); + + impl crate::connector::ConnectorBuilder for OwningConnectorBuilder { + fn build<'a>( + &'a self, + _db: &'a crate::AimDb, + ) -> Pin< + Box< + dyn Future< + Output = DbResult + Send + 'static>>>>, + > + Send + + 'a, + >, + > { + Box::pin(async { Ok(Vec::new()) }) + } + fn scheme(&self) -> &str { + self.0 + } + fn owns_scheme(&self) -> bool { + true + } + } + + /// Two scheme-owning connectors under one scheme is a configuration error, + /// not a silently duplicated route set. + #[tokio::test] + async fn two_owning_connectors_on_one_scheme_fail_the_build() { + let builder = crate::AimDbBuilder::new() + .runtime(Arc::new(MockRuntime)) + .with_connector(OwningConnectorBuilder("knx")) + .with_connector(OwningConnectorBuilder("knx")); + + let Err(err) = builder.build().await else { + panic!("a second connector for an owned scheme must be rejected"); + }; + let msg = alloc::format!("{err}"); + assert!( + msg.contains("More than one connector registered for scheme 'knx'"), + "unexpected error: {msg}" + ); + } + + /// Distinct schemes are fine — that is how a caller runs two of the same + /// transport side by side. + #[tokio::test] + async fn owning_connectors_on_distinct_schemes_build() { + let builder = crate::AimDbBuilder::new() + .runtime(Arc::new(MockRuntime)) + .with_connector(OwningConnectorBuilder("knx")) + .with_connector(OwningConnectorBuilder("mqtt")); + + assert!(builder.build().await.is_ok()); + } + + /// A connector that collects no routes — a session *server*, say — may be + /// registered twice under one scheme: two endpoints onto one dispatch. + /// `owns_scheme` defaults to `false`, so this must keep working. + #[tokio::test] + async fn two_non_owning_connectors_on_one_scheme_build() { + let builder = crate::AimDbBuilder::new() + .runtime(Arc::new(MockRuntime)) + .with_connector(NoopConnectorBuilder) + .with_connector(NoopConnectorBuilder); + + assert!(builder.build().await.is_ok()); + } + /// Acceptance criterion: a builder with three distinct /// mistakes reports all three from one `build()` call. #[tokio::test] diff --git a/aimdb-embassy-adapter/CHANGELOG.md b/aimdb-embassy-adapter/CHANGELOG.md index 65ecc83d..04140067 100644 --- a/aimdb-embassy-adapter/CHANGELOG.md +++ b/aimdb-embassy-adapter/CHANGELOG.md @@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **`EmbassyUdpBinder::bind` reports `TransportError::Busy`, not `Io`, when its + socket is already held.** The binder owns exactly one socket and `Clone` + shares it, so a clone binding while another handle holds it failed as a + generic `Io` — indistinguishable from a real bind failure, which a consumer's + retry loop treats as transient. A caller mistake then read as an endless + unexplained bind failure. `EmbassyTcpDialer` already returned `Busy` for the + identical case; the UDP binder now matches it, and its docs say (as the TCP + dialer's do) that `Clone` shares the socket, why the bound exists, and that a + second concurrent socket needs another `EmbassyNet::udp` call with its own + buffers — which matters because that constructor serves both KNX/IP and SNTP. + `tests/udp.rs` pins the variant (it previously asserted only `is_err()`, so it + passed either way) and adds the clone case the `Clone` derive invites, + including that releasing the socket frees the clone. + ### Changed (breaking) - **Issue #131 — `EmbassyAdapter` is a stateless unit type; network capability moves to connector construction.** The `EmbassyNetwork` trait and `EmbassyAdapter::new_with_network` are deleted (an `Arc` runtime can't surface adapter-specific capabilities); network connectors take the `embassy_net::Stack` at construction, wrapped in the new force-`Send + Sync` `connectors::NetStack` so the single-core `unsafe` stays in the audited `connectors` module — the adapter itself now carries **zero `unsafe`**. `EmbassyAdapter::new()` returns `Self` (was a never-failing `ExecutorResult` forcing `.unwrap()` at every call site) and `new_db_result()` is deleted. `NetStack::new` is an `unsafe fn`: the force-`Send + Sync` rests on the single-core cooperative-executor invariant, which the constructor cannot check, so each connector constructing one acknowledges it with a `SAFETY` comment (constructing on a multicore / multi-executor setup is UB). `EmbassyRecordRegistrarExt` shrinks to `.buffer(cfg)`; `EmbassyRecordRegistrarExtCustom` (`buffer_sized`, `source_with_context`) re-targets the non-generic `RecordRegistrar<'a, T>` with the concrete `RuntimeContext`, and `source_with_context` drops its needless `Sync` bounds (`Ctx: Send`, `F: Send`, matching core's relaxed `source`). `join_queue.rs` (`EmbassyJoinQueue`) is deleted with the `JoinFanInRuntime` family; the core join queue closes when forwarders exit (the Embassy queue previously never closed) and its capacity is 16 (was 8). diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index e947a34a..4bd0f247 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -442,6 +442,21 @@ impl Datagram for EmbassyUdpSocket { } /// Binds [`EmbassyUdpSocket`]s over one caller-owned socket. +/// +/// `Clone` shares that socket rather than duplicating it — it exists so a +/// binder can satisfy the `Clone` bound a connector needs to hand an owned +/// binder to its `'static` task (`KnxConnector`, for one). A clone binding +/// while another handle holds the socket gets [`TransportError::Busy`]. For a +/// second *concurrent* socket call [`EmbassyNet::udp`] again with its own +/// buffers, which is the only way to get one — `EmbassyNet::udp` serves both +/// KNX/IP and SNTP, so two consumers on one stack need two calls, not two +/// clones. +/// +/// Note this is unlike [`TokioUdpBinder`](https://docs.rs/aimdb-tokio-adapter), +/// whose `bind` opens a fresh OS socket per call and whose clones are therefore +/// independent. The same `Clone` bound means different things on the two +/// adapters, which is why it is spelled out here. +#[derive(Clone)] pub struct EmbassyUdpBinder { stack: Stack<'static>, slot: Arc, @@ -457,12 +472,17 @@ impl DatagramBinder for EmbassyUdpBinder { fn bind(&self, port: u16) -> impl Future> + Send + '_ { SendFutureWrapper(async move { + // `Busy`, not `Io`: the socket is held by another handle on this + // binder (a clone, or a live `EmbassyUdpSocket` not yet dropped). + // That is a caller mistake and never resolves on its own, where a + // failed `bind` below may; a retry loop that cannot tell them apart + // spins forever on the first with nothing to point at. let mut socket = self .slot .socket .borrow_mut() .take() - .ok_or(TransportError::Io)?; + .ok_or(TransportError::Busy)?; // Idempotent: a socket returned by a dropped `EmbassyUdpSocket` is // already closed, and closing an unbound socket is a no-op. socket.close(); diff --git a/aimdb-embassy-adapter/tests/udp.rs b/aimdb-embassy-adapter/tests/udp.rs index 1701c9d5..e2245be2 100644 --- a/aimdb-embassy-adapter/tests/udp.rs +++ b/aimdb-embassy-adapter/tests/udp.rs @@ -10,7 +10,7 @@ extern crate alloc; use core::future::Future; -use aimdb_core::session::{Datagram, DatagramBinder}; +use aimdb_core::session::{Datagram, DatagramBinder, TransportError}; use aimdb_embassy_adapter::net::EmbassyNet; use embassy_net::udp::PacketMetadata; use embassy_net::{Config, Ipv4Address, Ipv4Cidr, Stack, StaticConfigV4}; @@ -207,13 +207,50 @@ fn a_binder_rebinds_after_its_socket_is_dropped() { } /// The binder owns exactly one socket, so a second bind while the first is -/// live is refused rather than silently sharing. +/// live is refused rather than silently sharing — as `Busy`, not `Io`. +/// +/// The variant is the point, not just the failure: a consumer's bind-retry +/// loop treats `Io` as transient, so reporting it for a socket that is held +/// turns a caller mistake into an endless unexplained retry. #[test] -fn a_second_bind_fails_while_the_socket_is_held() { +fn a_second_bind_is_busy_while_the_socket_is_held() { let outcome = drive(|a_stack, _b_stack| async move { let binder = EmbassyNet::udp(a_stack, meta(), buf(), meta(), buf()); let _held = binder.bind(3671).await.expect("first bind"); - assert!(binder.bind(3672).await.is_err(), "socket is already taken"); + assert_eq!( + binder.bind(3672).await.err(), + Some(TransportError::Busy), + "the socket is held, so the second bind is Busy, not Io" + ); + }); + assert_eq!(outcome, Ok(())); +} + +/// A clone shares the one socket rather than duplicating it, so it is `Busy` +/// too — the case the `Clone` derive invites, since a connector needs the +/// bound to hand an owned binder to its `'static` task. +/// +/// Releasing the socket frees the clone: the sharing is a live handoff, not a +/// permanent claim by whoever bound first. +#[test] +fn a_clone_shares_the_socket_and_is_busy_until_it_is_released() { + let outcome = drive(|a_stack, _b_stack| async move { + let binder = EmbassyNet::udp(a_stack, meta(), buf(), meta(), buf()); + let clone = binder.clone(); + + let held = binder.bind(3671).await.expect("first bind"); + assert_eq!( + clone.bind(3672).await.err(), + Some(TransportError::Busy), + "a clone shares the socket, so its bind is Busy" + ); + + drop(held); + let recovered = clone + .bind(3672) + .await + .expect("the clone binds once released"); + assert_eq!(recovered.local_addr().unwrap().port(), 3672); }); assert_eq!(outcome, Ok(())); } diff --git a/aimdb-knx-connector/CHANGELOG.md b/aimdb-knx-connector/CHANGELOG.md index 12816867..f5633ede 100644 --- a/aimdb-knx-connector/CHANGELOG.md +++ b/aimdb-knx-connector/CHANGELOG.md @@ -9,9 +9,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- **`tokio-runtime` gains `embassy-sync`; `embassy-futures` is unconditional.** - Both are executor-independent, so one channel and select type serves either - runtime. +- **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. +- **Feature gates split `std` / `no_std + alloc` instead of naming runtimes + (breaking), matching the TCP and serial connectors.** The new `connector` + gate is the whole connector — tunnel engine, connection task, `KnxConnector` + — on `no_std + alloc`; `std = ["connector", "aimdb-core/std", + "knx-pico/std"]` only lifts `no_std` and adds knx-pico's std error impls plus + the back-compat DPT re-exports. Neither gate names an executor, so a + FreeRTOS/lwIP caller enables `connector` and passes its own binder and clock + without claiming to be Embassy. `tokio-runtime` and `embassy-runtime` remain + as deprecated aliases for `std` and `connector`; remove after a release. + + This drops the dependencies the deleted per-runtime clients had needed: + `aimdb-embassy-adapter`, `embassy-net`, `embassy-time`, `static_cell` and + `dep:defmt` (plus the long-unused `futures-core`, `thiserror` and + `embassy-executor`). The embedded graph goes from 88 crates to 46, and the + crate no longer depends on *either* adapter — the asymmetry that survived the + constructor change, since only the Tokio side had been cleaned up. + + `embassy-sync` stays, moved onto `connector`: it is no_std, no_alloc, pulls + no executor, and its `Channel` is this connector's public queue type + (`Channels`) on every runtime. `embassy-futures` stays unconditional. - **Selecting a `critical-section` implementation is left to the final binary.** `CriticalSectionRawMutex` needs one to link, but the impl is registered by symbol name and is global to the binary, so a library that enables it hands @@ -32,8 +59,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 over core's `DatagramBinder` and `Delay`, it binds, advertises the socket's real local endpoint (HPAI) instead of the NAT-style `0.0.0.0:0` — falling back to NAT on a cycle whose stack exposes no address, so a rebind never re-advertises the - previous cycle's port — drives the shared `TunnelEngine`, and rebinds on socket - reset. Its select alternates the inbound and command arms each pass, since + previous cycle's port, and on a socket bound to an unspecified IP (see *Fixed*) + — drives the shared `TunnelEngine`, and rebinds on socket reset. Its select alternates the inbound and command arms each pass, since `select3` polls in declaration order where the `tokio::select!` it replaces chose among ready arms at random. `shared_channel` bridges it to the `embassy_sync` channels, which now back the task on std too. @@ -41,6 +68,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **A `Busy` bind is logged as the caller mistake it is.** The bind-retry loop + treated every failure alike, so a binder whose socket is held elsewhere — a + clone of a single-socket `EmbassyUdpBinder`, say — produced one identical + "KNX bind failed; retrying" line every 5 s with nothing pointing at the + cause. `TransportError::Busy` now logs distinctly. Recovery is unchanged: a + `Busy` binder can free up if the other holder drops, so both cases still + retry. +- **A second KNX connector now fails the build instead of silently duplicating + every route.** `KnxConnector` declares core's new + `ConnectorBuilder::owns_scheme`, so registering two is a configuration error. + It never worked: `build` collects every `knx://` route regardless of which + gateway it was meant for, so each connector claimed all of them — every + `link_to` got two publishers, and sharing one `Channels` additionally put two + connection tasks on one command queue, splitting writes between gateways at + random. This is a guard, not a new limit: one connector is one tunnel to one + gateway and carries the whole bus behind it, with as many group addresses as + records declare. What it rules out is a second *gateway*, which the + scheme-keyed routing cannot express. +- **An unspecified bind address now advertises the NAT HPAI, not a real port + beside `0.0.0.0`.** The CONNECT_REQUEST's HPAI was built from `local_addr()` + with no check on the address, so binding `0.0.0.0` — the host default, and + what both demos, the crate doc example and `aimdb-codegen` all pass, since a + binary rarely knows which interface to pick — put `0.0.0.0:` on the + wire. That is neither an endpoint a gateway can reach nor the NAT form the + spec defines, and a gateway honouring the HPAI would send its tunnel data into + the void. An unspecified IP now takes the `LocalEndpoint::Nat` branch and + emits `0.0.0.0:0` (KNXnet/IP 5.2.3), which tells the gateway to reply to the + datagram's source address. A bind to a real interface address is unaffected + and still advertised explicitly. Both pre-rewrite clients had this (the tokio + one bound `"0.0.0.0:0"` and ran the same logic), so the handshake changes for + every default deployment. - **Inbound single-octet telegrams no longer decode to `0` (#210).** A telegram carrying exactly one data octet — every DPT5 datapoint (5.001 percentage, 5.003 angle, 5.010 counter, …) — was published as `0` instead of its value. `knx-pico` derived the application data as `[9 .. 7 + npdu_length)`, one octet short of the KNX encoding (the NPDU length octet counts the APCI octet plus the data octets, so the data spans `[9 .. 8 + npdu_length)`); the slice came back empty, the telegram was taken for a 6-bit encoded one, and its value was read out of the APCI octet as `0x80 & 0x3F` — zero. Only single-octet payloads were affected: DPT1 was genuinely 6-bit encoded, and DPT9/DPT14 happened to work because the old code ignored the parsed slice and read to the end of the datagram. Fixed at the root in the fork (`aimdb-dev/knx-pico` `b4883c4`, reported upstream as [cc90202/knx-pico#4](https://github.com/cc90202/knx-pico/issues/4)) — the same off-by-one that made 6-bit telegrams panic, which the previously carried patch had only clamped to an empty slice. `parse_telegram` now reads the parsed frame instead of re-deriving cEMI offsets, so the payload is bounded by the NPDU length octet rather than running to the end of the datagram. **Requires the updated fork** — see the patch note in the [usage guide](../docs/aimdb-usage-guide.md). - **Heartbeat-response liveness — a dead send path or expired gateway channel now reconnects (review follow-up to #135).** The engine tracks each CONNECTIONSTATE_REQUEST and drops the connection when the gateway's CONNECTIONSTATE_RESPONSE doesn't arrive within the new `TunnelConfig::heartbeat_response_timeout_ms` (default 10 s, the KNX spec timeout) or reports a non-zero status (e.g. the gateway expired the channel during an outage). This restores the old tokio client's recovery from silently-failing sends — the recv path of an unconnected UDP socket never errors, so without it a route flap left the tunnel `Connected` forever with a stale channel id — and adds genuine liveness detection on both runtimes. - **Pending-ACK tracking is accurate under send failures and bursts.** A frame the transport could not hand to the socket is untracked (`TunnelIo::send` reports success; previously the 3 s sweep warned "ACK timeout" for a telegram that never left the host), and a burst deeper than the 16-entry pending map evicts-and-reports the oldest entry instead of silently dropping its timeout reporting. diff --git a/aimdb-knx-connector/Cargo.toml b/aimdb-knx-connector/Cargo.toml index 52c7b130..8d6e9ca1 100644 --- a/aimdb-knx-connector/Cargo.toml +++ b/aimdb-knx-connector/Cargo.toml @@ -13,16 +13,34 @@ 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", + +# The whole connector: the sans-io tunnel engine, the unified connection task, +# and `KnxConnector` over core's `DatagramBinder`/`Delay`. `src/` names no +# adapter and no `std::`, so one gate covers every target. A runtime is chosen +# by passing an adapter's binder and clock — a dependency the caller adds, not a +# 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. +connector = [ + "aimdb-core/alloc", + "aimdb-core/connector-session", # `pump_sink`/`pump_source`/`Source`/`Payload` "embassy-sync", ] +# Orthogonal to `connector`: the code is `alloc`-only either way, so this only +# lifts `no_std`, forwards core's own `std`, and adds knx-pico's std error impls +# plus the back-compat DPT re-exports. +std = ["connector", "aimdb-core/std", "knx-pico/std"] + +# Deprecated aliases. They name executors this crate never mentions, which +# blocks a FreeRTOS/lwIP build from enabling the connector without claiming to +# be Embassy. Kept so existing consumers keep working; remove after a release. +tokio-runtime = ["std"] +embassy-runtime = ["connector"] + # Selects `critical-section`'s std implementation. # # `embassy-sync`'s only `Sync` raw mutex is `CriticalSectionRawMutex`, which @@ -31,92 +49,46 @@ tokio-runtime = [ # 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 `tokio-runtime`. This crate's own +# 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 -# documented way) or enable this feature. +# documented way) or enable this feature. On Embassy the HAL (cortex-m / +# embassy-rp / ...) already provides one. critical-section-std-impl = ["critical-section/std"] -embassy-runtime = [ - "aimdb-core/alloc", # Need alloc for collect_inbound_routes - "aimdb-core/connector-session", # `pump_sink`/`pump_source`/`Source`/`Payload` - "dep:aimdb-embassy-adapter", # Enable the optional dependency - "aimdb-embassy-adapter/embassy-net-support", # Enable EmbassyNetwork trait for network stack access - "aimdb-embassy-adapter/connectors", # `into_box_future` spine helper - "embassy-executor", - "embassy-time", - "embassy-sync", - "embassy-net", - "static_cell", -] + # Design 050 §10.4/§10.5: the facade reaches both destinations through # `aimdb_core::__private`, so neither dependency is declared here any more. # The *features* stay: a `#[cfg]` in a `#[macro_export]`ed macro is resolved # where it expands, so without them this crate would emit nothing. tracing = ["aimdb-core/tracing"] log = ["aimdb-core/log"] -defmt = [ - "dep:defmt", - "aimdb-core/defmt", - "knx-pico/defmt", -] # Only use knx-pico's defmt for logging +defmt = ["aimdb-core/defmt", "knx-pico/defmt"] [dependencies] aimdb-core = { version = "1.1.0", path = "../aimdb-core", default-features = false } -aimdb-embassy-adapter = { version = "0.6.0", path = "../aimdb-embassy-adapter", default-features = false, optional = true } # aimdb-dev fork: upstream 0.3.0 panics on an npdu_length=1 telegram. A patch # won't do — patches aren't published, so dependants resolve back to upstream. # Keyed `knx-pico` to keep imports and feature references unchanged. knx-pico = { package = "aimdb-knx-pico", version = "0.3.1", default-features = false } -# 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) -# Note: These use the workspace's local embassy checkout to avoid conflicts -embassy-executor = { version = "0.10.0", optional = true } -embassy-time = { version = "0.5.1", optional = true } +# Executor-independent, despite the names: neither pulls an executor and both +# build on std, so one channel type and one select serve every runtime. +# `embassy-sync` is optional because it is only reachable through `connector`; +# `embassy-futures` is unconditional (its `[dependencies]` is empty bar optional +# defmt/log, so the std graph is unaffected). embassy-sync = { version = "0.8.0", path = "../_external/embassy/embassy-sync", optional = true } -# Unconditional: its `[dependencies]` is empty bar optional defmt/log, so the -# std graph is unaffected and one select loop serves both runtimes. embassy-futures = { version = "0.1.2" } -embassy-net = { version = "0.9.0", optional = true, features = [ - "tcp", - "udp", - "dhcpv4", - "medium-ethernet", - "proto-ipv4", -] } # A `critical-section` impl must be linked wherever `CriticalSectionRawMutex` # is used. Choosing one is the final binary's call, so it is reachable only -# through the opt-in `critical-section-std-impl` feature above; on Embassy the -# HAL (cortex-m / embassy-rp / …) already provides it. +# through the opt-in `critical-section-std-impl` feature above. critical-section = { version = "1.1", optional = true } # Embedded utilities (heapless is unconditional: the shared sans-io tunnel # engine uses stack-allocated frames on both runtimes) heapless = { workspace = true } -static_cell = { version = "2.0", optional = true } - -# Optional observability -defmt = { workspace = true, optional = true } [dev-dependencies] tokio = { workspace = true, features = ["full"] } diff --git a/aimdb-knx-connector/README.md b/aimdb-knx-connector/README.md index ccf139af..78add83b 100644 --- a/aimdb-knx-connector/README.md +++ b/aimdb-knx-connector/README.md @@ -10,7 +10,14 @@ Add to your `Cargo.toml`: ```toml [dependencies] -aimdb-knx-connector = { version = "0.1", features = ["tokio-runtime"] } +# `std` is the host leg. `critical-section-std-impl` selects the impl the +# connector's channels need to link — only a final binary may pick one. +aimdb-knx-connector = { version = "0.5", features = [ + "std", + "critical-section-std-impl", +] } +# The host also needs the adapter's UDP socket and clock. +aimdb-tokio-adapter = { version = "0.6", features = ["tokio-runtime", "net"] } # REQUIRED: Patch knx-pico to use fork with bug fixes [patch.crates-io] @@ -26,7 +33,8 @@ We're working with upstream to get these changes merged. Once published, the pat ## Features -- **Dual Runtime Support**: Works with both Tokio (std) and Embassy (no_std) runtimes +- **Runtime-Neutral**: One connector for every runtime — you pass an adapter's + UDP binder and clock, so no feature here names an executor - **KNXnet/IP Tunneling**: Full protocol support via UDP port 3671 - **Bidirectional Communication**: Monitor bus activity and send commands - **Type-Safe Records**: KNX telegrams become strongly-typed Rust records @@ -37,19 +45,33 @@ We're working with upstream to get these changes merged. Once published, the pat ## Quick Start (Tokio) ```rust -use aimdb_knx_connector::KnxConnector; +use aimdb_knx_connector::{Channels, KnxConnector}; +use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; use aimdb_tokio_adapter::TokioAdapter; +use std::net::Ipv4Addr; #[derive(Debug, Clone)] 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() .runtime(TokioAdapter::new()?) - .with_connector(KnxConnector::new("knx://192.168.1.19:3671")) + // The adapter owns the UDP socket and the clock; this crate owns the + // tunnelling protocol. + .with_connector(KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, + "knx://192.168.1.19:3671", + &CHANNELS, + )) .configure::(|reg| { reg.buffer(BufferCfg::SingleLatest) .link_from("knx://1/0/7") @@ -74,7 +96,30 @@ async fn main() -> Result<(), Box> { ## Quick Start (Embassy) -See `examples/embassy-knx-connector-demo/` for embedded usage. +The same constructor — only the binder and clock change, and the channels come +from a `static` instead of being sized at runtime: + +```rust +use aimdb_embassy_adapter::net::{EmbassyDelay, EmbassyNet}; +use aimdb_knx_connector::{Channels, KnxConnector}; + +static CHANNELS: Channels<32> = Channels::new(); + +let binder = EmbassyNet::udp(stack, rx_meta, rx_buf, tx_meta, tx_buf); + +let builder = AimDbBuilder::new() + .runtime(runtime) + .with_connector(KnxConnector::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 +`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 d7da1a47..6b01384e 100644 --- a/aimdb-knx-connector/src/client.rs +++ b/aimdb-knx-connector/src/client.rs @@ -15,7 +15,9 @@ use core::future::Future; use core::net::SocketAddr; use core::time::Duration; -use aimdb_core::session::{Datagram, DatagramBinder, Delay, Payload, TransportResult}; +use aimdb_core::session::{ + Datagram, DatagramBinder, Delay, Payload, TransportError, TransportResult, +}; use aimdb_core::{log_debug, log_error, log_warn, RuntimeOps}; use crate::tunnel::{ @@ -209,6 +211,20 @@ pub async fn connection_task( loop { let mut socket = match binder.bind(0).await { Ok(socket) => socket, + // `Busy` is a caller mistake, not a transient fault: the binder's + // one socket is held elsewhere — typically a clone of a + // single-socket binder — and no amount of retrying frees it. Same + // recovery either way (a `Busy` binder *can* free up if the other + // holder drops), but the log has to say which, or the misuse reads + // as an endless unexplained bind failure. + Err(TransportError::Busy) => { + log_error!( + "KNX bind failed: the binder's socket is held by another handle \ + (a clone of a single-socket binder?); retrying" + ); + delay.sleep(BIND_RETRY).await; + continue; + } Err(_) => { log_error!("KNX bind failed; retrying"); delay.sleep(BIND_RETRY).await; @@ -225,8 +241,17 @@ pub async fn connection_task( // is exactly what causes a rebind. Leaving the previous cycle's value // in place would advertise a port nothing is bound to any more and wedge // the handshake for good; NAT is degraded but recovers. + // + // An unspecified IP is NAT too, not an address. Binding `0.0.0.0` is the + // normal host default (it is what the demos, the doc example and + // `aimdb-codegen` all pass), and the socket then reports `0.0.0.0:port` + // — a real port paired with an IP that routes nowhere. Advertising that + // verbatim is worse than either honest option: a gateway that honours + // the HPAI sends its tunnel data into the void, while `0.0.0.0:0` is the + // form KNXnet/IP 5.2.3 defines for exactly this case and makes the + // gateway reply to the datagram's source address instead. match socket.local_addr() { - Some(SocketAddr::V4(addr)) => { + Some(SocketAddr::V4(addr)) if !addr.ip().is_unspecified() => { engine.set_local_endpoint(LocalEndpoint::Explicit { ip: addr.ip().octets(), port: addr.port(), @@ -257,7 +282,7 @@ 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(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] pub mod shared_channel { use super::{CommandSource, GroupWrite, Payload, TelegramSink}; use alloc::string::String; @@ -288,7 +313,7 @@ pub mod shared_channel { } } -#[cfg(all(test, feature = "tokio-runtime"))] +#[cfg(all(test, feature = "std"))] mod tests { use super::*; use aimdb_core::session::TransportError; @@ -384,6 +409,48 @@ mod tests { task.abort(); } + /// The counterpart of the test above, for the bind address everything + /// actually ships with. + /// + /// `Ipv4Addr::UNSPECIFIED` is what the demos, the crate doc example and + /// `aimdb-codegen` all pass, so `local_addr()` reports `0.0.0.0:`. + /// That must go out as the NAT HPAI (`0.0.0.0:0`), not as the port paired + /// with an IP that routes nowhere — a gateway honouring the latter would + /// send its tunnel data into the void. + #[tokio::test] + async fn an_unspecified_bind_address_advertises_the_nat_hpai() { + let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("bind fake gateway"); + let gateway_addr = gateway.local_addr().expect("gateway addr"); + + let task = tokio::spawn(connection_task( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + gateway_addr, + runtime(), + TokioDelay, + VecSink::default(), + NoCommands, + )); + + let mut buf = [0u8; 128]; + let (len, _from) = tokio::time::timeout(RECV_TIMEOUT, gateway.recv_from(&mut buf)) + .await + .expect("gateway received no CONNECT_REQUEST") + .expect("recv_from"); + + assert!(len >= 14, "CONNECT_REQUEST should carry both HPAIs"); + assert_eq!(&buf[8..12], &[0, 0, 0, 0], "NAT HPAI address"); + assert_eq!( + u16::from_be_bytes([buf[12], buf[13]]), + 0, + "NAT HPAI must zero the port too: a real port beside 0.0.0.0 is \ + neither an endpoint nor the spec's NAT form" + ); + + 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. @@ -540,6 +607,78 @@ mod tests { } } + /// Reports `Busy` for the first two binds, then hands over a real socket — + /// a single-socket binder whose socket another handle is holding. + struct BusyThenFreeBinder { + attempts: Arc, + } + + impl DatagramBinder for BusyThenFreeBinder { + type Socket = FlappingSocket; + + async fn bind(&self, port: u16) -> TransportResult { + if self.attempts.fetch_add(1, Ordering::SeqCst) < 2 { + return Err(TransportError::Busy); + } + let inner = tokio::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, port)) + .await + .map_err(|_| TransportError::Io)?; + Ok(FlappingSocket { + inner, + report_addr: true, + fail_recv: false, + }) + } + } + + /// A `Busy` bind must not end the task: it retries, and connects once the + /// other handle releases the socket. + /// + /// `start_paused` so the `BIND_RETRY` sleeps are virtual — the two retries + /// cost 10 s of tokio's clock and no wall time. + #[tokio::test(start_paused = true)] + async fn a_busy_binder_retries_and_recovers() { + let gateway = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("bind fake gateway"); + let gateway_addr = gateway.local_addr().expect("gateway addr"); + let attempts = Arc::new(AtomicUsize::new(0)); + + let task = tokio::spawn(connection_task( + BusyThenFreeBinder { + attempts: attempts.clone(), + }, + gateway_addr, + runtime(), + TokioDelay, + VecSink::default(), + NoCommands, + )); + + // Must exceed the two `BIND_RETRY` sleeps the task waits out. `RECV_TIMEOUT` + // would not: it and the first retry share the t=5s deadline, and under a + // paused clock the timeout wins before the second retry is ever reached. + const PAST_TWO_RETRIES: std::time::Duration = std::time::Duration::from_secs(60); + + let mut buf = [0u8; 128]; + let (len, _) = tokio::time::timeout(PAST_TWO_RETRIES, gateway.recv_from(&mut buf)) + .await + .expect("no CONNECT_REQUEST: the task did not retry past Busy") + .expect("recv_from"); + + assert!( + len >= 14, + "CONNECT_REQUEST reached the wire after the retries" + ); + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "two Busy binds, then the one that succeeded" + ); + + task.abort(); + } + /// A rebind that cannot learn its address must advertise the NAT-style /// HPAI, never the previous cycle's port. /// diff --git a/aimdb-knx-connector/src/connector.rs b/aimdb-knx-connector/src/connector.rs new file mode 100644 index 00000000..df9df0bf --- /dev/null +++ b/aimdb-knx-connector/src/connector.rs @@ -0,0 +1,425 @@ +//! 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, ChannelSink}, +}; +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 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(), + ChannelSink::(channels.telegrams.sender()), + 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" + } + + /// One KNX connector per db. + /// + /// Not a limit on group addresses — one connector is one tunnel to one + /// gateway, and that is the whole bus behind it: every record's + /// `link_from`/`link_to` names its own address, and they all ride this one + /// connector. What it rules out is a *second gateway*, which cannot work + /// today because `build` collects every `knx://` route regardless of which + /// gateway it was meant for. + fn owns_scheme(&self) -> bool { + true + } +} + +#[cfg(all(test, feature = "std"))] +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}" + ); + } + + /// A db with one inbound and one outbound `knx` route, so both pumps have + /// something to contribute. + /// + /// 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. + 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( + TokioNet::udp(Ipv4Addr::LOCALHOST), + TokioDelay, + "knx://127.0.0.1:3671", + ®ISTERED, + )); + builder.configure::("switch", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_from("knx://1/0/7") + .with_deserializer( + |_ctx, data: &[u8]| Ok(data.first().copied().unwrap_or(0) as u64), + ) + .finish(); + }); + builder.configure::("lamp", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .link_to("knx://1/0/6") + .with_serializer(|_ctx, v: &u64| Ok(vec![*v as u8])) + .finish(); + }); + builder.build().await.expect("build db").0 + } + + /// 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. + #[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, + ); + + let futures = connector.build(&db).await.expect("build"); + assert_eq!( + futures.len(), + 2, + "connection task + pump_source, and no publisher" + ); + } + + /// 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. + #[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 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}" + ); + } + } + + /// The whole wiring against a real UDP gateway: the task binds, advertises + /// its endpoint, and the handshake reaches the wire. + /// + /// Binds `LOCALHOST`, not the `UNSPECIFIED` the demos and `aimdb-codegen` + /// pass, precisely so `local_addr()` yields a routable address and the + /// explicit-HPAI branch is the one under test. The NAT branch that an + /// 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( + 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..c717b5e7 100644 --- a/aimdb-knx-connector/src/lib.rs +++ b/aimdb-knx-connector/src/lib.rs @@ -6,11 +6,25 @@ //! //! ## Features //! -//! - `tokio-runtime`: Tokio-based connector using UDP sockets -//! - `embassy-runtime`: Embassy connector for embedded systems +//! No feature here names a runtime: `src/` mentions no adapter and no `std::`, +//! and a runtime is chosen by *passing* an adapter's `DatagramBinder` and +//! `Delay` to `KnxConnector::new`. A third runtime (FreeRTOS/lwIP) works with +//! no edit to this crate. +//! +//! - `connector`: the whole connector — tunnel engine, connection task, and +//! `KnxConnector` — on `no_std + alloc`. This is the gate an embedded caller +//! enables. +//! - `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. //! - `tracing`: Debug logging support (std) //! - `defmt`: Debug logging support (no_std) //! +//! `tokio-runtime` and `embassy-runtime` are deprecated aliases for `std` and +//! `connector` respectively, kept for one release. +//! //! ## Production Status //! //! **Current Version: 0.1.0 - Beta Quality** @@ -34,8 +48,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 +62,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 +100,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) @@ -149,34 +171,14 @@ pub use knx_pico::dpt::{Dpt1, Dpt5, Dpt9, DptDecode, DptEncode}; // Runtime-neutral KNX/IP tunneling state machine shared by both transports. pub mod tunnel; -// The connection task: one body for both runtimes, generic over core's -// datagram and delay traits. Supersedes the two per-runtime client modules -// below, which it will replace outright. -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +// The connection task: one body for every runtime, generic over core's +// datagram and delay traits. +#[cfg(feature = "connector")] 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(feature = "connector")] +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(feature = "connector")] +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/src/tunnel.rs b/aimdb-knx-connector/src/tunnel.rs index bf7fae4d..ba04cae5 100644 --- a/aimdb-knx-connector/src/tunnel.rs +++ b/aimdb-knx-connector/src/tunnel.rs @@ -528,11 +528,8 @@ impl TunnelEngine { /// non-blocking inbound forward. Implemented once per runtime shim; the policy /// for applying [`Action`]s lives in [`drain_actions`] so it cannot drift /// between the tokio and Embassy loops. -// Unused only when the crate is built without either runtime shim. -#[cfg_attr( - not(any(feature = "tokio-runtime", feature = "embassy-runtime")), - allow(dead_code) -)] +// Unused only when the crate is built without the connection task. +#[cfg_attr(not(feature = "connector"), allow(dead_code))] pub(crate) trait TunnelIo { /// Send one datagram to the gateway. Returns `false` when the datagram /// could not be handed to the socket; [`drain_actions`] then stops the @@ -558,10 +555,7 @@ pub(crate) trait TunnelIo { /// Apply every pending engine action through `io`. Returns `true` when the /// engine asked for the socket to be torn down ([`Action::ResetSocket`]); the /// transport reacts once the drain completes. -#[cfg_attr( - not(any(feature = "tokio-runtime", feature = "embassy-runtime")), - allow(dead_code) -)] +#[cfg_attr(not(feature = "connector"), allow(dead_code))] pub(crate) async fn drain_actions(engine: &mut TunnelEngine, io: &mut impl TunnelIo) -> bool { let mut reset = false; while let Some(action) = engine.next_action() { diff --git a/aimdb-knx-connector/tests/connection_state_tests.rs b/aimdb-knx-connector/tests/connection_state_tests.rs index 2eb3d9de..88f64d59 100644 --- a/aimdb-knx-connector/tests/connection_state_tests.rs +++ b/aimdb-knx-connector/tests/connection_state_tests.rs @@ -1,6 +1,6 @@ //! Integration tests for connection state management -#[cfg(feature = "tokio-runtime")] +#[cfg(feature = "std")] mod tests { #[test] fn test_channel_state_sequence_management() { diff --git a/aimdb-knx-connector/tests/frame_building_tests.rs b/aimdb-knx-connector/tests/frame_building_tests.rs index c6e8749d..5a1cb888 100644 --- a/aimdb-knx-connector/tests/frame_building_tests.rs +++ b/aimdb-knx-connector/tests/frame_building_tests.rs @@ -1,6 +1,6 @@ //! Unit tests for KNX frame building and parsing -#[cfg(feature = "tokio-runtime")] +#[cfg(feature = "std")] mod tests { #[test] fn test_connect_request_structure() { diff --git a/aimdb-knx-connector/tests/group_address_tests.rs b/aimdb-knx-connector/tests/group_address_tests.rs index d3269273..4f4bccd8 100644 --- a/aimdb-knx-connector/tests/group_address_tests.rs +++ b/aimdb-knx-connector/tests/group_address_tests.rs @@ -1,6 +1,6 @@ //! Unit tests for KNX group address parsing and formatting -#[cfg(feature = "tokio-runtime")] +#[cfg(feature = "std")] mod tokio_tests { use aimdb_knx_connector::GroupAddress; diff --git a/aimdb-knx-connector/tests/shared_channel_on_std.rs b/aimdb-knx-connector/tests/shared_channel_on_std.rs index f0e986e5..c4bcbd24 100644 --- a/aimdb-knx-connector/tests/shared_channel_on_std.rs +++ b/aimdb-knx-connector/tests/shared_channel_on_std.rs @@ -8,7 +8,7 @@ //! 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 = "tokio-runtime")] +#![cfg(feature = "std")] use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex; use embassy_sync::channel::Channel; diff --git a/aimdb-knx-connector/tests/topic_provider_tests.rs b/aimdb-knx-connector/tests/topic_provider_tests.rs index f5f2f8c1..2441dfa7 100644 --- a/aimdb-knx-connector/tests/topic_provider_tests.rs +++ b/aimdb-knx-connector/tests/topic_provider_tests.rs @@ -6,15 +6,22 @@ //! //! The tests use mock data and don't require a running KNX/IP gateway. -#![cfg(feature = "tokio-runtime")] +#![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 // ============================================================================ @@ -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 afe5454f..265e0703 100644 --- a/aimdb-tokio-adapter/src/net.rs +++ b/aimdb-tokio-adapter/src/net.rs @@ -151,6 +151,7 @@ impl Datagram for TokioDatagram { } /// Binds [`TokioDatagram`]s, one per reconnect cycle. +#[derive(Clone, Copy)] pub struct TokioUdpBinder { local_ip: IpAddr, } diff --git a/docs/aimdb-usage-guide.md b/docs/aimdb-usage-guide.md index c69c6b58..f5dee285 100644 --- a/docs/aimdb-usage-guide.md +++ b/docs/aimdb-usage-guide.md @@ -47,17 +47,18 @@ edition = "2021" [dependencies] # AimDB core and Tokio runtime adapter -aimdb-core = "0.3" -aimdb-tokio-adapter = { version = "0.3", features = ["tokio-runtime"] } +aimdb-core = "1.2" +aimdb-tokio-adapter = { version = "0.6", features = ["tokio-runtime"] } # Optional: KNX connector -# aimdb-knx-connector = { version = "0.2", features = ["tokio-runtime"] } +# aimdb-knx-connector = { version = "0.5", features = ["std", "critical-section-std-impl"] } +# aimdb-tokio-adapter's "net" feature is also required — see the KNX section. # Optional: MQTT connector -# aimdb-mqtt-connector = { version = "0.3", features = ["tokio-runtime"] } +# aimdb-mqtt-connector = { version = "0.6", features = ["tokio-runtime"] } # Optional: Sync API wrapper (for blocking code) -# aimdb-sync = "0.3" +# aimdb-sync = "0.6" # Tokio runtime tokio = { version = "1.0", features = ["full"] } @@ -148,18 +149,18 @@ Once you have an Embassy project set up, add AimDB: ```toml [dependencies] # AimDB core and Embassy runtime adapter -aimdb-core = { version = "0.3", default-features = false } -aimdb-embassy-adapter = { version = "0.3", features = ["embassy-runtime", "embassy-task-pool-16"] } +aimdb-core = { version = "1.2", default-features = false } +aimdb-embassy-adapter = { version = "0.6", features = ["embassy-runtime"] } # Optional: KNX connector -# aimdb-knx-connector = { version = "0.2", features = ["embassy-runtime"], default-features = false } +# aimdb-knx-connector = { version = "0.5", features = ["connector"], default-features = false } # Optional: MQTT connector -# aimdb-mqtt-connector = { version = "0.3", features = ["embassy-runtime"], default-features = false } +# aimdb-mqtt-connector = { version = "0.6", features = ["embassy-runtime"], default-features = false } # Embassy runtime (example for RP2040) -embassy-executor = { version = "0.6", features = ["arch-cortex-m", "executor-thread"] } -embassy-time = { version = "0.3", features = ["generic-queue-16"] } # REQUIRED — see note below +embassy-executor = { version = "0.10", features = ["platform-cortex-m", "executor-thread"] } +embassy-time = { version = "0.5", features = ["generic-queue-16"] } # REQUIRED — see note below embassy-rp = { version = "0.2", features = ["time-driver"] } # CRITICAL: Patch dependencies for compatibility @@ -295,11 +296,14 @@ The KNX connector provides KNX/IP tunneling support for building automation syst **Add to Cargo.toml:** ```toml -# For Tokio -aimdb-knx-connector = { version = "0.2", features = ["tokio-runtime"] } +# Host. `critical-section-std-impl` selects the impl the connector's channels +# need to link — only a final binary may pick one. +aimdb-knx-connector = { version = "0.5", features = ["std", "critical-section-std-impl"] } +aimdb-tokio-adapter = { version = "0.6", features = ["tokio-runtime", "net"] } -# For Embassy -aimdb-knx-connector = { version = "0.2", features = ["embassy-runtime"], default-features = false } +# Embedded. `connector` is the same code on `no_std + alloc`; the HAL already +# provides a `critical-section` impl. +aimdb-knx-connector = { version = "0.5", features = ["connector"], default-features = false } # REQUIRED PATCH (bug fixes not yet on crates.io) [patch.crates-io] @@ -310,7 +314,9 @@ knx-pico = { git = "https://github.com/aimdb-dev/knx-pico.git", branch = "master ```rust use aimdb_core::prelude::*; use aimdb_tokio_adapter::TokioAdapter; -use aimdb_knx_connector::KnxConnector; +use aimdb_tokio_adapter::net::{TokioDelay, TokioNet}; +use aimdb_knx_connector::{Channels, KnxConnector}; +use std::net::Ipv4Addr; use std::sync::Arc; #[derive(Debug, Clone)] @@ -318,13 +324,23 @@ 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()?); let db = AimDbBuilder::new() .runtime(runtime) - .with_connector(KnxConnector::new("knx://192.168.1.19:3671")) + // The adapter owns the UDP socket and the clock; the connector owns + // the tunnelling protocol. + .with_connector(KnxConnector::new( + TokioNet::udp(Ipv4Addr::UNSPECIFIED), + TokioDelay, + "knx://192.168.1.19:3671", + &CHANNELS, + )) .configure::("light.state", |reg| { reg.buffer(BufferCfg::SingleLatest) .link_from("knx://1/0/7") @@ -358,10 +374,10 @@ The MQTT connector enables pub/sub messaging with MQTT brokers. **Add to Cargo.toml:** ```toml # For Tokio -aimdb-mqtt-connector = { version = "0.3", features = ["tokio-runtime"] } +aimdb-mqtt-connector = { version = "0.6", features = ["tokio-runtime"] } # For Embassy -aimdb-mqtt-connector = { version = "0.3", features = ["embassy-runtime"], default-features = false } +aimdb-mqtt-connector = { version = "0.6", features = ["embassy-runtime"], default-features = false } # REQUIRED PATCH for Embassy (version compatibility) [patch.crates-io] @@ -538,17 +554,17 @@ For maximum stability, you can pin to specific versions on crates.io: ```toml [dependencies] -aimdb-core = "=0.3.0" -aimdb-tokio-adapter = "=0.3.0" -aimdb-mqtt-connector = "=0.3.0" -aimdb-knx-connector = "=0.2.0" +aimdb-core = "=1.2.0" +aimdb-tokio-adapter = "=0.6.0" +aimdb-mqtt-connector = "=0.6.0" +aimdb-knx-connector = "=0.5.0" ``` Or use standard semver: ```toml [dependencies] -aimdb-core = "0.3" # Will use latest 0.3.x -aimdb-knx-connector = "0.2" # Will use latest 0.2.x +aimdb-core = "1.2" # Will use latest 1.2.x +aimdb-knx-connector = "0.5" # Will use latest 0.5.x ``` ## Migration Path @@ -558,14 +574,14 @@ As bug fixes are upstreamed and published, the patches can be removed: ```toml # Current (with patches) [dependencies] -aimdb-knx-connector = "0.2" +aimdb-knx-connector = "0.5" [patch.crates-io] knx-pico = { git = "https://github.com/aimdb-dev/knx-pico.git", branch = "master" } # After upstream fixes are published [dependencies] -aimdb-knx-connector = "0.2" # or newer version +aimdb-knx-connector = "0.5" # or newer version # [patch.crates-io] <-- Just delete this section! ``` diff --git a/examples/embassy-knx-connector-demo/Cargo.toml b/examples/embassy-knx-connector-demo/Cargo.toml index 7a75b043..86632730 100644 --- a/examples/embassy-knx-connector-demo/Cargo.toml +++ b/examples/embassy-knx-connector-demo/Cargo.toml @@ -19,9 +19,15 @@ aimdb-embassy-adapter = { path = "../../aimdb-embassy-adapter", default-features "embassy-runtime", # `EmbassyUart`, the byte source the serial connector frames. "connector-io", + # `EmbassyNet::udp` and `EmbassyDelay`, the UDP socket and clock the KNX + # connector tunnels over. Until #251 this arrived transitively through the + # serial connector; that crate is runtime-neutral now, so declare it here. + "net", ] } aimdb-knx-connector = { path = "../../aimdb-knx-connector", default-features = false, features = [ - "embassy-runtime", + # No runtime named: the connector is neutral and takes this binary's + # `EmbassyNet`/`EmbassyDelay` as arguments. + "connector", "defmt", ] } # Serial remote-access server — serves this db's records over a UART (ST-LINK VCP) diff --git a/examples/embassy-knx-connector-demo/src/main.rs b/examples/embassy-knx-connector-demo/src/main.rs index aee7ad7e..cbddd5be 100644 --- a/examples/embassy-knx-connector-demo/src/main.rs +++ b/examples/embassy-knx-connector-demo/src/main.rs @@ -43,13 +43,15 @@ extern crate alloc; use aimdb_core::remote::SecurityPolicy; 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::dpt::{Dpt1, Dpt9, DptDecode, DptEncode}; -use aimdb_knx_connector::embassy_client::KnxConnectorBuilder; use aimdb_serial_connector::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}; @@ -270,11 +272,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(EmbassyUart::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..59f39adc 100644 --- a/examples/tokio-knx-connector-demo/Cargo.toml +++ b/examples/tokio-knx-connector-demo/Cargo.toml @@ -7,8 +7,7 @@ description = "AimDB example demonstrating KNX connector integration with Tokio publish = false [features] -default = ["tokio-runtime", "tracing"] -tokio-runtime = ["aimdb-knx-connector/tokio-runtime"] +default = ["tracing"] tracing = ["dep:tracing", "dep:tracing-subscriber"] [dependencies] @@ -16,6 +15,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", ] } @@ -27,7 +28,11 @@ knx-connector-demo-common = { path = "../knx-connector-demo-common", features = # KNX connector aimdb-knx-connector = { path = "../../aimdb-knx-connector", features = [ - "tokio-runtime", + # The host leg: the runtime-neutral connector plus core's `std`. + "std", + # 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/README.md b/examples/tokio-knx-connector-demo/README.md index be290d0d..3f5460c1 100644 --- a/examples/tokio-knx-connector-demo/README.md +++ b/examples/tokio-knx-connector-demo/README.md @@ -27,9 +27,16 @@ Demonstrates bidirectional KNX/IP integration with AimDB using the Tokio runtime Edit `src/main.rs` to match your KNX setup: ```rust -// Gateway URL -.with_connector(aimdb_knx_connector::KnxConnector::new( +// 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 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));