diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9b317425..05eb0051 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -165,3 +165,10 @@ jobs: - name: Generate documentation run: make doc + + # `make check` runs every crate-level leg but not `examples`, so the demo + # binaries — the only end-to-end proof of the embedded path — were the one + # artifact CI never compiled. `all` is `build test examples`; the first two + # re-run here against a warm cache. + - name: Build everything, including the example binaries + run: make all diff --git a/Cargo.lock b/Cargo.lock index 4437ede8..c4dec9cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -354,7 +354,6 @@ dependencies = [ "futures", "serde", "serde_json", - "thiserror 2.0.17", "tokio", "tokio-serial", ] diff --git a/Makefile b/Makefile index 97110e57..462ebf7a 100644 --- a/Makefile +++ b/Makefile @@ -124,8 +124,10 @@ build: cargo build --package aimdb-websocket-connector --features "server,client" @printf "$(YELLOW) → Building UDS connector$(NC)\n" cargo build --package aimdb-uds-connector + @printf "$(YELLOW) → Building serial connector (neutral: framer + sugar, no adapter)$(NC)\n" + cargo build --package aimdb-serial-connector --no-default-features --features "connector" @printf "$(YELLOW) → Building serial connector (tokio)$(NC)\n" - cargo build --package aimdb-serial-connector --no-default-features --features "tokio-runtime" + cargo build --package aimdb-serial-connector --no-default-features --features "std" @printf "$(YELLOW) → Building TCP connector (tokio)$(NC)\n" cargo build --package aimdb-tcp-connector --no-default-features --features "tokio-runtime" @printf "$(YELLOW) → Building WASM adapter$(NC)\n" @@ -218,9 +220,9 @@ test: @printf "$(YELLOW) → Testing UDS connector$(NC)\n" cargo test --package aimdb-uds-connector @printf "$(YELLOW) → Testing serial connector (tokio: COBS framing + AimX round-trip over a duplex)$(NC)\n" - cargo test --package aimdb-serial-connector --no-default-features --features "_test-tokio" + cargo test --package aimdb-serial-connector --no-default-features --features "std" @printf "$(YELLOW) → Testing serial connector (embassy: COBS framing + client-engine smoke on the EmbassyAdapter clock)$(NC)\n" - cargo test --package aimdb-serial-connector --no-default-features --features "embassy-runtime" + cargo test --package aimdb-serial-connector --no-default-features --features "_test-embassy" @printf "$(YELLOW) → Testing TCP connector (tokio: length-prefix framing + AimX loopback)$(NC)\n" cargo test --package aimdb-tcp-connector --no-default-features --features "_test-tokio" @printf "$(YELLOW) → Testing TCP connector (embassy: socket recycle + concurrent slots + redial over an embassy-net loopback)$(NC)\n" @@ -341,11 +343,11 @@ clippy: @printf "$(YELLOW) → Clippy on UDS connector$(NC)\n" cargo clippy --package aimdb-uds-connector --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on serial connector (tokio)$(NC)\n" - cargo clippy --package aimdb-serial-connector --no-default-features --features "_test-tokio" --all-targets -- -D warnings + cargo clippy --package aimdb-serial-connector --no-default-features --features "std" --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on serial connector (embassy)$(NC)\n" - cargo clippy --package aimdb-serial-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime" -- -D warnings + cargo clippy --package aimdb-serial-connector --target thumbv7em-none-eabihf --no-default-features --features "_test-embassy" -- -D warnings @printf "$(YELLOW) → Clippy on serial connector (embassy + defmt)$(NC)\n" - cargo clippy --package aimdb-serial-connector --target thumbv7em-none-eabihf --no-default-features --features "embassy-runtime,defmt" -- -D warnings + cargo clippy --package aimdb-serial-connector --target thumbv7em-none-eabihf --no-default-features --features "_test-embassy,defmt" -- -D warnings @printf "$(YELLOW) → Clippy on TCP connector (tokio)$(NC)\n" cargo clippy --package aimdb-tcp-connector --no-default-features --features "_test-tokio" --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on TCP connector (embassy)$(NC)\n" @@ -382,12 +384,23 @@ doc: cargo doc --package aimdb-persistence --no-deps cargo doc --package aimdb-persistence-sqlite --no-deps cargo doc --package aimdb-websocket-connector --features "tokio-runtime" --no-deps + cargo doc --package aimdb-uds-connector --no-deps + @# The serial and TCP connectors document a different item set per feature: + @# `std` adds the host-only port/socket helpers, `connector` is the neutral + @# half an MCU gets. Both legs run - a link from an ungated item to a gated + @# one only breaks in the leg that lacks it (see the embedded section below). + cargo doc --package aimdb-serial-connector --no-default-features --features "std" --no-deps + cargo doc --package aimdb-tcp-connector --no-default-features --features "std" --no-deps + cargo doc --package aimdb-client --features "transport-serial,transport-tcp" --no-deps + cargo doc --package aimdb-derive --no-deps @cp -r target/doc/* target/doc-final/cloud/ @printf "$(YELLOW) → Building embedded documentation$(NC)\n" 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-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/ @printf "$(YELLOW) → Building WASM/browser documentation$(NC)\n" cargo doc --package aimdb-wasm-adapter --target wasm32-unknown-unknown --features "wasm-runtime" --no-deps @@ -456,9 +469,9 @@ test-embedded: @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-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 "embassy-runtime" + 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" - cargo check --package aimdb-serial-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime,defmt" + cargo check --package aimdb-serial-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "_test-embassy,defmt" @printf "$(YELLOW) → Checking aimdb-tcp-connector (Embassy TCP client) on thumbv7em-none-eabihf target$(NC)\n" cargo check --package aimdb-tcp-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime" @printf "$(YELLOW) → Checking aimdb-tcp-connector (Embassy TCP client + defmt) on thumbv7em-none-eabihf target$(NC)\n" diff --git a/aimdb-client/Cargo.toml b/aimdb-client/Cargo.toml index 1dc61f25..6548c24c 100644 --- a/aimdb-client/Cargo.toml +++ b/aimdb-client/Cargo.toml @@ -35,7 +35,7 @@ aimdb-core = { version = "1.1.0", path = "../aimdb-core", features = [ # binary links only what it needs. aimdb-uds-connector = { version = "0.1.0", path = "../aimdb-uds-connector", optional = true } aimdb-serial-connector = { version = "0.1.0", path = "../aimdb-serial-connector", default-features = false, features = [ - "tokio-runtime", + "std", ], optional = true } aimdb-tcp-connector = { version = "0.1.0", path = "../aimdb-tcp-connector", default-features = false, features = [ "tokio-runtime", diff --git a/aimdb-client/src/endpoint.rs b/aimdb-client/src/endpoint.rs index b12e3dbc..c2b00f9a 100644 --- a/aimdb-client/src/endpoint.rs +++ b/aimdb-client/src/endpoint.rs @@ -131,7 +131,7 @@ pub fn dial(endpoint: &str) -> ClientResult> { Scheme::Serial => { #[cfg(feature = "transport-serial")] { - Ok(Box::new(aimdb_serial_connector::SerialDialer::new( + Ok(Box::new(aimdb_serial_connector::SerialPortDialer::new( parsed.target, parsed.baud.unwrap_or(DEFAULT_SERIAL_BAUD), ))) diff --git a/aimdb-core/src/session/connector.rs b/aimdb-core/src/session/connector.rs index f2ddc098..febdd205 100644 --- a/aimdb-core/src/session/connector.rs +++ b/aimdb-core/src/session/connector.rs @@ -29,9 +29,9 @@ use crate::builder::AimDb; use crate::connector::ConnectorBuilder; use crate::session::{ pump_client, run_client, serve, ClientConfig, Dialer, Dispatch, EnvelopeCodec, Listener, - SessionConfig, + OneShot, SessionConfig, }; -use crate::DbResult; +use crate::{DbError, DbResult}; /// The default scheme a session connector registers when none is given. pub const DEFAULT_SCHEME: &str = "remote"; @@ -48,7 +48,7 @@ type BuildFuture<'a> = Pin>> + S /// it in a one-line sugar constructor (e.g. `UdsClient`). pub struct SessionClientConnector { scheme: String, - dialer: D, + dialer: OneShot, codec: C, config: ClientConfig, } @@ -59,7 +59,7 @@ impl SessionClientConnector { pub fn new(dialer: D, codec: C) -> Self { Self { scheme: DEFAULT_SCHEME.to_string(), - dialer, + dialer: OneShot::new(dialer), codec, config: ClientConfig::default(), } @@ -81,13 +81,22 @@ impl SessionClientConnector { impl ConnectorBuilder for SessionClientConnector where - D: Dialer + Clone + Send + Sync + 'static, + D: Dialer + Send + 'static, C: EnvelopeCodec + Clone + 'static, { fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { Box::pin(async move { + // Taken on first poll, not at call time: a `build()` future dropped + // before it is polled must leave the dialer where it was. + let dialer = self + .dialer + .take() + .ok_or_else(|| DbError::InvalidOperation { + operation: "SessionClientConnector::build".to_string(), + reason: "the moved-in dialer was already taken; build() ran twice".to_string(), + })?; let (handle, engine_fut) = run_client( - self.dialer.clone(), + dialer, self.codec.clone(), self.config.clone(), db.runtime_ops(), diff --git a/aimdb-core/src/session/io.rs b/aimdb-core/src/session/io.rs index 999c8595..531e2dd2 100644 --- a/aimdb-core/src/session/io.rs +++ b/aimdb-core/src/session/io.rs @@ -414,6 +414,65 @@ impl From for OneShot { } } +/// Hands out one pre-built connection, then refuses. +/// +/// The [`Dialer`] a point-to-point transport wants: a moved-in UART, pipe or +/// socket has nothing to redial, so the second attempt is +/// [`TransportError::Closed`] — which [`run_client`](super::run_client) treats +/// as terminal — rather than a silent reconnect loop. +pub struct OneShotDialer { + conn: OneShot, +} + +impl OneShotDialer { + /// Hold `conn` for the first [`connect`](Dialer::connect). + pub fn new(conn: C) -> Self { + Self { + conn: OneShot::new(conn), + } + } +} + +impl Dialer for OneShotDialer { + fn connect(&self) -> BoxFut<'_, TransportResult>> { + Box::pin(async move { + self.conn + .take() + .map(|c| Box::new(c) as Box) + .ok_or(TransportError::Closed) + }) + } +} + +/// Hands out one pre-built connection, then parks forever. +/// +/// The [`Listener`] dual of [`OneShotDialer`]: [`serve`](super::serve) loops on +/// `accept`, and a point-to-point link has no second peer, so parking is the +/// correct end state rather than an error the loop would spin on. +pub struct OneShotListener { + conn: OneShot, +} + +impl OneShotListener { + /// Hold `conn` for the first [`accept`](Listener::accept). + pub fn new(conn: C) -> Self { + Self { + conn: OneShot::new(conn), + } + } +} + +impl Listener for OneShotListener { + fn accept(&mut self) -> BoxFut<'_, TransportResult>> { + Box::pin(async move { + match self.conn.take() { + Some(c) => Ok(Box::new(c) as Box), + None => core::future::pending().await, + } + }) + } +} + // =========================================================================== // Compile-time assertions. // =========================================================================== diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs index 85c5f170..55b839f8 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -50,7 +50,8 @@ pub use endpoint::{split_host_port, split_host_port_opt, EndpointError}; #[cfg(feature = "connector-session")] pub use io::{ ByteStream, Datagram, DatagramBinder, Delay, FrameFault, FramedConnection, Framer, - FramerFactory, FramingDialer, FramingListener, IoError, OneShot, StreamDialer, StreamListener, + FramerFactory, FramingDialer, FramingListener, IoError, OneShot, OneShotDialer, + OneShotListener, StreamDialer, StreamListener, }; #[cfg(feature = "connector-session")] pub use pump::{pump_sink, pump_source}; diff --git a/aimdb-core/tests/session_engine.rs b/aimdb-core/tests/session_engine.rs index 94799f24..2521c9ef 100644 --- a/aimdb-core/tests/session_engine.rs +++ b/aimdb-core/tests/session_engine.rs @@ -19,8 +19,9 @@ use futures::StreamExt; use aimdb_core::session::{ run_client, serve, AuthError, BoxFut, BoxStream, ClientConfig, CodecError, Connection, Dialer, - Dispatch, EnvelopeCodec, Inbound, Listener, Outbound, Payload, PeerInfo, RpcError, Session, - SessionConfig, SessionCtx, SessionLimits, SubUpdate, TransportError, TransportResult, + Dispatch, EnvelopeCodec, Inbound, Listener, OneShotDialer, OneShotListener, Outbound, Payload, + PeerInfo, RpcError, Session, SessionConfig, SessionCtx, SessionLimits, SubUpdate, + TransportError, TransportResult, }; /// Engine-test clock (aimdb-core can't depend on a runtime adapter — that @@ -1005,3 +1006,95 @@ async fn a_concurrent_batch_of_timeouts_leaves_the_connection_usable() { drop(peer); let _ = client.await; } + +// =========================================================================== +// One-shot transports — the moved-in dual of the re-dialable/re-bindable pair. +// +// Ported here from `aimdb-embassy-adapter`'s connector smoke test when the +// Embassy-specific spine was retired: the types are runtime-neutral now, and +// `serve` has no other test of its own. +// =========================================================================== + +/// A moved-in stream has nothing to redial, so the second attempt must be an +/// error — and specifically `Closed`, which `run_client` treats as terminal +/// rather than backing off into a permanently-failing redial loop. +#[tokio::test] +async fn a_one_shot_dialer_hands_out_one_connection_then_reports_closed() { + let (a, _b) = conn_pair(); + let dialer = OneShotDialer::new(a); + + assert!(dialer.connect().await.is_ok(), "first connect hands it out"); + assert_eq!( + dialer.connect().await.err(), + Some(TransportError::Closed), + "`Closed` is what stops the engine; `Io` would earn a backoff and retry" + ); +} + +/// The listener dual: `serve` loops on `accept`, so a point-to-point link with +/// no second peer must park rather than error — an erroring accept would tear +/// the server down. +#[tokio::test] +async fn a_one_shot_listener_parks_after_its_only_accept() { + let (a, _b) = conn_pair(); + let mut listener = OneShotListener::new(a); + + assert!(listener.accept().await.is_ok(), "first accept yields"); + assert!( + tokio::time::timeout(Duration::from_millis(50), listener.accept()) + .await + .is_err(), + "the second accept must park, not resolve" + ); +} + +/// End to end: `serve` handles the one peer the listener has, and when that +/// peer hangs up it goes back to waiting on the parked accept instead of +/// returning — the exact shape a UART server runs in. +#[tokio::test] +async fn serve_over_a_one_shot_listener_handles_the_peer_then_keeps_waiting() { + let writes: WriteLog = Arc::new(Mutex::new(Vec::new())); + let dispatch = Arc::new(EchoDispatch { + writes: writes.clone(), + }); + + let (server_end, mut client_end) = conn_pair(); + let server = tokio::spawn(serve( + OneShotListener::new(server_end), + Arc::new(LineCodec), + dispatch, + SessionConfig::default(), + )); + + // One fire-and-forget write, then hang up. + client_end + .send( + b"WRITE +topic +hello", + ) + .await + .expect("send the write"); + drop(client_end); + + // The session ran: the write reached the dispatch. + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + loop { + if !writes.lock().unwrap().is_empty() { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "the write never reached the dispatch" + ); + tokio::task::yield_now().await; + } + + // ...and `serve` is still running, parked on the second accept. + assert!( + tokio::time::timeout(Duration::from_millis(100), server) + .await + .is_err(), + "serve must keep waiting on the parked accept, not return" + ); +} diff --git a/aimdb-embassy-adapter/Cargo.toml b/aimdb-embassy-adapter/Cargo.toml index 10e4fdda..a38a999e 100644 --- a/aimdb-embassy-adapter/Cargo.toml +++ b/aimdb-embassy-adapter/Cargo.toml @@ -22,17 +22,20 @@ embassy-runtime = ["embassy-executor", "embassy-time", "embassy-sync"] embassy-net-support = ["embassy-net"] # Network stack support for connectors connectors = ["aimdb-core/connector-session"] +# The framed `Connection` spine plus the non-socket byte sources in `io` +# (`EmbassyUart`): core's I/O traits over `embedded-io-async`, no network stack. connector-io = ["connectors", "dep:embedded-io-async"] -# Embassy sockets, dialers and UART halves behind core's runtime-neutral I/O -# traits, so connector crates need no `embassy-net` dependency. Deliberately +# Embassy sockets and dialers behind core's runtime-neutral I/O traits, so +# connector crates need no `embassy-net` dependency. Deliberately # does not imply `embassy-time`: that turns on `defmt-timestamp-uptime`, whose # `_defmt_timestamp` collides with every host test binary's `defmt::timestamp!`. net = [ - "connectors", + # The UART in `io` is `connector-io`'s, not this feature's; `net` enables it + # so one feature still gets a caller every byte source the adapter offers. + "connector-io", "embassy-net-support", "embassy-net/udp", - "dep:embedded-io-async", "dep:embassy-futures", ] diff --git a/aimdb-embassy-adapter/src/connectors.rs b/aimdb-embassy-adapter/src/connectors.rs index 94e51c65..899cfd9f 100644 --- a/aimdb-embassy-adapter/src/connectors.rs +++ b/aimdb-embassy-adapter/src/connectors.rs @@ -1,65 +1,49 @@ -//! Centralized Embassy connector spines — the one audited home for the -//! single-core `unsafe` + [`SendFutureWrapper`] that -//! every Embassy connector used to hand-roll. +//! The data-plane bridge — the one audited home for the single-core `unsafe` + +//! [`SendFutureWrapper`] that every Embassy data-plane connector used to +//! hand-roll. //! //! AimDB's connector contract is `Send`-everywhere (so a Tokio app can -//! `tokio::spawn(runner.run())`). Embassy's primitives (UART halves over -//! `embedded-io-async`, channels over `NoopRawMutex`, …) are `!Send` *by design* -//! — single-core, cooperative, no preemption or thread migration. Bridging the -//! two requires force-`Send`ing the Embassy futures; this module does that -//! **once**, so a connector crate carries **no `unsafe` and no wrapper**: +//! `tokio::spawn(runner.run())`). Embassy's primitives (channels over +//! `NoopRawMutex`, a borrowed `embassy_net::Stack`, …) are `!Send` *by design* — +//! single-core, cooperative, no preemption or thread migration. Bridging the two +//! requires force-`Send`ing the Embassy futures; this module does that **once**, +//! so a connector crate carries **no `unsafe` and no wrapper**. //! -//! - **Session transports** (serial, TCP, …) contribute a `Framer` -//! (`connector-io` feature) or a [`Connection`], and wrap it in -//! [`EmbassySessionClient`] / -//! [`EmbassySessionServer`] — the Embassy duals of core's -//! `SessionClientConnector` / `SessionServerConnector`. -//! - **Data-plane transports** (MQTT, KNX) contribute an [`EmbassySinkRaw`] -//! (outbound) and/or [`EmbassySourceRaw`] (inbound) and ride core's existing -//! [`pump_sink`](aimdb_core::session::pump_sink) / -//! [`pump_source`](aimdb_core::session::pump_source) via the force-`Send` -//! bridges [`EmbassySink`] / [`EmbassySource`]. +//! Data-plane transports (MQTT, KNX) contribute an [`EmbassySinkRaw`] (outbound) +//! and/or [`EmbassySourceRaw`] (inbound) and ride core's +//! [`pump_sink`](aimdb_core::session::pump_sink) / +//! [`pump_source`](aimdb_core::session::pump_source) via the force-`Send` +//! bridges [`EmbassySink`] / [`EmbassySource`]. +//! +//! Session transports (serial, TCP, …) no longer come through here. They ride +//! core's runtime-neutral spine directly — `SessionClientConnector` / +//! `SessionServerConnector` over `FramedConnection`, with the byte source from +//! this crate's `io` or `net` module (unlinked: neither exists in a +//! `connectors`-only build) — so the Embassy duals this module used to +//! carry (`EmbassySessionClient`/`Server`, `EmbassyConnection`, `OneShotCell` +//! and the one-shot dialer/listener) are gone. Their one-shot semantics live in +//! core as `OneShot`, `OneShotDialer` and `OneShotListener`. //! //! # Safety invariant (shared by every `unsafe impl` below) //! //! An Embassy executor runs cooperatively on a single core with no preemption or //! thread migration, so the wrapped `!Send` values are never actually accessed -//! from another thread. Only use these spines under an Embassy executor. +//! from another thread. Only use these bridges under an Embassy executor. -use core::cell::RefCell; use core::future::Future; use core::pin::Pin; use alloc::boxed::Box; use alloc::string::{String, ToString}; -use alloc::sync::Arc; -use alloc::vec; use alloc::vec::Vec; -use aimdb_core::connector::ConnectorBuilder; -use aimdb_core::session::{ - pump_client, run_client, serve, BoxFut, ClientConfig, Connection, Dialer, Dispatch, - EnvelopeCodec, Listener, Payload, SessionConfig, Source, TransportError, TransportResult, -}; +use aimdb_core::session::{BoxFut, Payload, Source}; use aimdb_core::transport::{Connector, ConnectorConfig, PublishError}; -use aimdb_core::{AimDb, DbError, DbResult}; use crate::SendFutureWrapper; -/// The scheme a spine registers when the connector gives none (matches core's -/// `SessionClientConnector` default). -pub const DEFAULT_SCHEME: &str = "remote"; - /// The runner's collected future type (`Send`, as the std contract requires). type BoxFuture = Pin + Send + 'static>>; -/// The `ConnectorBuilder::build` return shape. -type BuildFuture<'a> = Pin>> + Send + 'a>>; - -/// The spine's one-shot peripheral was already consumed — `build` ran twice. The -/// framework calls it once, so this is unreachable in practice. -fn connector_consumed() -> DbError { - DbError::missing_configuration("embassy connector already built") -} // =========================================================================== // Data-plane bridges — let a `!Send` sink/source ride core's pumps. @@ -144,7 +128,7 @@ where /// Force-`Send + Sync` handle to the Embassy network stack. /// /// `embassy_net::Stack` is `!Sync` (internal `RefCell`), so a -/// [`ConnectorBuilder`] (which must be `Send + Sync`) cannot hold the bare +/// `ConnectorBuilder` (which must be `Send + Sync`) cannot hold the bare /// `&'static Stack`. Network connectors (MQTT, KNX) take the stack at /// construction and wrap it here — keeping the single-core `unsafe` in this /// audited module instead of in every connector crate. Replaces the deleted @@ -183,333 +167,3 @@ impl NetStack { self.0 } } - -// =========================================================================== -// Session spine — the Embassy duals of `SessionClientConnector` / `…Server`. -// =========================================================================== - -/// A force-`Send + Sync` one-shot cell. Holds a moved-in value behind interior -/// mutability so a [`ConnectorBuilder`] (which is `Send + Sync`) can take it once -/// from `&self` in `build` — without the connector crate writing any `unsafe`. -/// -/// Use it when a session-server connector holds a moved-in peripheral/connection -/// it hands to a [`OneShotListener`] at build time (the moved-in dual of the -/// Tokio server's re-bindable listener factory). -pub struct OneShotCell { - inner: RefCell>, -} - -// SAFETY: single-core cooperative Embassy executor — see the module-level invariant. -unsafe impl Send for OneShotCell {} -// SAFETY: same invariant; the `RefCell` is never borrowed from another thread. -unsafe impl Sync for OneShotCell {} - -impl OneShotCell { - /// Hold `value` for a single [`take`](Self::take). - pub fn new(value: C) -> Self { - Self { - inner: RefCell::new(Some(value)), - } - } - - /// Take the value, or `None` if already taken (i.e. `build` ran twice). - pub fn take(&self) -> Option { - self.inner.borrow_mut().take() - } - - /// Take the value, or the canonical "already built" [`DbError`] — the shared - /// error every spine returns when `build` is (impossibly) called twice. - pub fn take_required(&self) -> DbResult { - self.take().ok_or_else(connector_consumed) - } -} - -/// One-shot [`Dialer`] over a pre-built, moved-in `Connection` (an Embassy -/// peripheral can't be re-acquired, so it dials exactly once; keep reconnect -/// disabled — [`EmbassySessionClient::new`]'s default). -pub struct OneShotDialer { - conn: OneShotCell, -} - -impl OneShotDialer { - /// Wrap a pre-built connection to be handed out on the first `connect`. - pub fn new(conn: C) -> Self { - Self { - conn: OneShotCell::new(conn), - } - } -} - -impl Dialer for OneShotDialer { - fn connect(&self) -> BoxFut<'_, TransportResult>> { - Box::pin(SendFutureWrapper(async move { - self.conn - .take() - .map(|c| Box::new(c) as Box) - .ok_or(TransportError::Io) - })) - } -} - -/// One-shot [`Listener`] over a pre-built, moved-in `Connection`: the first -/// `accept` hands it out; later calls park forever (point-to-point peripheral). -pub struct OneShotListener { - conn: Option, -} - -// SAFETY: single-core cooperative Embassy executor — see the module-level invariant. -unsafe impl Send for OneShotListener {} - -impl OneShotListener { - /// Wrap a pre-built connection to be handed out on the first `accept`. - pub fn new(conn: C) -> Self { - Self { conn: Some(conn) } - } -} - -impl Listener for OneShotListener { - fn accept(&mut self) -> BoxFut<'_, TransportResult>> { - Box::pin(SendFutureWrapper(async move { - match self.conn.take() { - Some(c) => Ok(Box::new(c) as Box), - // Point-to-point: no second peer ever arrives. - None => core::future::pending().await, - } - })) - } -} - -/// Embassy dual of `SessionClientConnector`: dials a peer with `D`, speaks codec -/// `C`, mirrors records under a [`scheme`](ConnectorBuilder::scheme). A transport -/// crate wraps it in a one-line sugar constructor (e.g. `SerialClient`). -pub struct EmbassySessionClient { - scheme: String, - // The moved-in dialer behind the force-`Send + Sync` cell, so the builder is - // auto `Send + Sync` (with a `Send + Sync` codec) — no `unsafe` here. - dialer: OneShotCell, - codec: C, - config: ClientConfig, -} - -impl EmbassySessionClient { - /// Mirror records over `dialer`, framing with `codec`. Scheme defaults to - /// [`DEFAULT_SCHEME`]. - /// - /// Reconnect is **disabled** by default (unlike [`ClientConfig::default`]): - /// an Embassy dialer typically wraps a moved-in peripheral ([`OneShotDialer`]) - /// that can't be re-acquired, so redialing would spin on [`TransportError::Io`] - /// forever. A transport whose dialer really can redial opts back in via - /// [`with_config`](Self::with_config). - pub fn new(dialer: D, codec: C) -> Self { - Self { - scheme: DEFAULT_SCHEME.to_string(), - dialer: OneShotCell::new(dialer), - codec, - config: ClientConfig { - reconnect: false, - ..ClientConfig::default() - }, - } - } - - /// Override the scheme this connector registers. - pub fn scheme(mut self, scheme: impl Into) -> Self { - self.scheme = scheme.into(); - self - } - - /// Override the client engine config (reconnect, keepalive, …). Only enable - /// `reconnect` if the dialer can actually redial (a [`OneShotDialer`] can't). - pub fn with_config(mut self, config: ClientConfig) -> Self { - self.config = config; - self - } -} - -impl ConnectorBuilder for EmbassySessionClient -where - D: Dialer + 'static, - C: EnvelopeCodec + Clone + 'static, -{ - fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { - Box::pin(SendFutureWrapper(async move { - let dialer = self.dialer.take_required()?; - let (handle, engine) = run_client( - dialer, - self.codec.clone(), - self.config.clone(), - db.runtime_ops(), - ); - // One pump future per route; each holds a `ClientHandle` clone, so the - // engine stays alive as long as any mirror runs. - let mut futures = pump_client(db, &self.scheme, &handle); - futures.push(engine); - Ok(futures) - })) - } - - fn scheme(&self) -> &str { - &self.scheme - } -} - -/// Embassy dual of `SessionServerConnector`: serves a moved-in [`Listener`] with -/// a dispatch produced from the live db, speaking codec `C`, under a -/// [`scheme`](ConnectorBuilder::scheme). -pub struct EmbassySessionServer { - scheme: String, - // Moved-in listener behind the force-`Send + Sync` cell, so the builder is - // auto `Send + Sync` (with a `Send + Sync` codec + factory) — no `unsafe` here. - listener: OneShotCell, - codec: C, - dispatch_factory: DF, - config: SessionConfig, -} - -impl EmbassySessionServer { - /// Serve `listener` with the dispatch built by `dispatch_factory` from the - /// live db, framing with `codec`. Scheme defaults to [`DEFAULT_SCHEME`]. - pub fn new(listener: L, codec: C, dispatch_factory: DF, config: SessionConfig) -> Self { - Self { - scheme: DEFAULT_SCHEME.to_string(), - listener: OneShotCell::new(listener), - codec, - dispatch_factory, - config, - } - } - - /// Override the scheme this connector registers. - pub fn scheme(mut self, scheme: impl Into) -> Self { - self.scheme = scheme.into(); - self - } -} - -impl ConnectorBuilder for EmbassySessionServer -where - L: Listener + 'static, - C: EnvelopeCodec + Clone + 'static, - DF: Fn(&AimDb) -> Arc + Send + Sync, -{ - fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { - Box::pin(SendFutureWrapper(async move { - let listener = self.listener.take_required()?; - let dispatch = (self.dispatch_factory)(db); - let codec = Arc::new(self.codec.clone()); - // `serve` is `Send` here because the listener/connections force-`Send` - // their futures; no extra wrapper needed. - let fut: BoxFuture = Box::pin(serve(listener, codec, dispatch, self.config.clone())); - Ok(vec![fut]) - })) - } - - fn scheme(&self) -> &str { - &self.scheme - } -} - -// =========================================================================== -// Framed connection over `embedded-io-async` — lets a session transport ship -// just a `Framer` and carry zero `unsafe`. -// =========================================================================== - -#[cfg(feature = "connector-io")] -mod framed { - use super::*; - use embedded_io_async::{Read, Write}; - - use aimdb_core::session::PeerInfo; - - /// Frames the byte stream a [`EmbassyConnection`] carries: a transport - /// contributes only this (e.g. COBS, length-prefix), inheriting the - /// force-`Send` plumbing. - pub trait Framer { - /// Encode one logical frame, appending its wire bytes to `out`. - fn encode(&self, frame: &[u8], out: &mut Vec); - /// Feed received bytes into the accumulator. - fn push_bytes(&mut self, bytes: &[u8]); - /// Pull the next complete frame: `Some(Ok(frame))`, `Some(Err(()))` for a - /// malformed/unsynced run (skipped, the stream resyncs), or `None`. - fn next_frame(&mut self) -> Option, ()>>; - } - - /// A framed bidirectional [`Connection`] over an `embedded-io-async` UART (or - /// any `Read`/`Write` halves), force-`Send`ing its `recv`/`send` futures. - /// `RC`/`WC` cap the per-`read`/`write` chunk (UART ring sizes). - pub struct EmbassyConnection { - rx: Rd, - tx: Wr, - framer: F, - peer: PeerInfo, - } - - // SAFETY: single-core cooperative Embassy executor — see the module-level invariant. - unsafe impl Send - for EmbassyConnection - { - } - - impl EmbassyConnection { - /// Wrap the split read/write halves of an async byte stream with `framer`. - pub fn new(rx: Rd, tx: Wr, framer: F) -> Self { - Self { - rx, - tx, - framer, - peer: PeerInfo::default(), - } - } - } - - impl Connection - for EmbassyConnection - where - Rd: Read, - Wr: Write, - F: Framer, - { - fn recv(&mut self) -> BoxFut<'_, TransportResult>>> { - Box::pin(SendFutureWrapper(async move { - loop { - // A run that fails to decode is line noise or a mid-stream - // join, not fatal: skip it and resync on the next frame. - match self.framer.next_frame() { - Some(Ok(frame)) => return Ok(Some(frame)), - Some(Err(())) => continue, - None => {} - } - let mut chunk = [0u8; RC]; - match self.rx.read(&mut chunk).await { - Ok(0) => return Ok(None), // EOF — peer closed - Ok(n) => self.framer.push_bytes(&chunk[..n]), - Err(_) => return Err(TransportError::Io), - } - } - })) - } - - fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> { - Box::pin(SendFutureWrapper(async move { - let mut out = Vec::new(); - self.framer.encode(frame, &mut out); - // Some HAL `BufferedUart::write` rejects a single write larger than - // its TX ring, so split into `WC`-sized chunks. - for chunk in out.chunks(WC) { - self.tx - .write_all(chunk) - .await - .map_err(|_| TransportError::Closed)?; - } - self.tx.flush().await.map_err(|_| TransportError::Closed) - })) - } - - fn peer(&self) -> &PeerInfo { - &self.peer - } - } -} - -#[cfg(feature = "connector-io")] -pub use framed::{EmbassyConnection, Framer}; diff --git a/aimdb-embassy-adapter/src/io.rs b/aimdb-embassy-adapter/src/io.rs new file mode 100644 index 00000000..90af083e --- /dev/null +++ b/aimdb-embassy-adapter/src/io.rs @@ -0,0 +1,201 @@ +//! Embassy byte sources behind core's runtime-neutral I/O traits, for the +//! transports that need no network stack. +//! +//! A UART is one of core's [`ByteStream`]s, but it borrows nothing from +//! `embassy-net`, so it lives here under `connector-io` rather than in the +//! `net` module under `net` — a board with no networking should not compile +//! `smoltcp` to frame a serial line. `net` enables `connector-io`, so the +//! socket types and these share one feature graph from the caller's side. +//! +//! # Safety invariant +//! +//! As in [`crate::connectors`]: an Embassy executor runs cooperatively on a +//! single core with no preemption or thread migration, so the wrapped `!Send` +//! values are never touched from another thread. The traits declare `+ Send` +//! futures and Embassy's are not, so each impl returns a +//! [`SendFutureWrapper`]. + +use core::future::Future; + +use aimdb_core::session::{ByteStream, TransportError, TransportResult}; + +use crate::SendFutureWrapper; + +/// A UART, or any `embedded-io-async` read/write pair, as one [`ByteStream`] — +/// so the connector names no `embedded-io-async` types of its own. +pub struct EmbassyUart { + rx: Rd, + tx: Wr, +} + +// SAFETY: single-core cooperative Embassy executor — see the module invariant. +unsafe impl Send for EmbassyUart {} + +impl EmbassyUart { + /// Present an already-split UART's halves as one stream. + pub fn new(rx: Rd, tx: Wr) -> Self { + Self { rx, tx } + } +} + +impl ByteStream for EmbassyUart +where + Rd: embedded_io_async::Read, + Wr: embedded_io_async::Write, +{ + fn read<'a>( + &'a mut self, + buf: &'a mut [u8], + ) -> impl Future> + Send + 'a { + SendFutureWrapper(async move { self.rx.read(buf).await.map_err(|_| TransportError::Io) }) + } + + fn write_all<'a>( + &'a mut self, + buf: &'a [u8], + ) -> impl Future> + Send + 'a { + SendFutureWrapper(async move { + self.tx + .write_all(buf) + .await + .map_err(|_| TransportError::Closed) + }) + } + + fn flush(&mut self) -> impl Future> + Send + '_ { + SendFutureWrapper(async move { self.tx.flush().await.map_err(|_| TransportError::Closed) }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aimdb_core::session::{Connection, FrameFault, FramedConnection, Framer}; + use alloc::vec; + use alloc::vec::Vec; + + // `host_test_stubs!()` must expand once per binary; `buffer` already does + // it for this one. + + /// An in-memory read half: hands out queued chunks, then EOF. + struct MockRx(Vec>); + + impl embedded_io_async::ErrorType for MockRx { + type Error = embedded_io_async::ErrorKind; + } + + impl embedded_io_async::Read for MockRx { + async fn read(&mut self, buf: &mut [u8]) -> Result { + if self.0.is_empty() { + return Ok(0); + } + let chunk = self.0.remove(0); + let n = chunk.len().min(buf.len()); + buf[..n].copy_from_slice(&chunk[..n]); + Ok(n) + } + } + + /// An in-memory write half recording what it was given. + #[derive(Default)] + struct MockTx { + written: Vec, + flushes: usize, + } + + impl embedded_io_async::ErrorType for MockTx { + type Error = embedded_io_async::ErrorKind; + } + + impl embedded_io_async::Write for MockTx { + async fn write(&mut self, buf: &[u8]) -> Result { + self.written.extend_from_slice(buf); + Ok(buf.len()) + } + async fn flush(&mut self) -> Result<(), Self::Error> { + self.flushes += 1; + Ok(()) + } + } + + /// Length-prefixed framer, enough to drive a `FramedConnection`. + #[derive(Default)] + struct LenFramer { + buf: Vec, + } + + impl Framer for LenFramer { + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { + out.push(frame.len() as u8); + out.extend_from_slice(frame); + Ok(()) + } + fn push_bytes(&mut self, bytes: &[u8]) { + self.buf.extend_from_slice(bytes); + } + fn next_frame(&mut self) -> Option, FrameFault>> { + let len = *self.buf.first()? as usize; + if self.buf.len() < len + 1 { + return None; + } + let frame = self.buf[1..len + 1].to_vec(); + self.buf.drain(..len + 1); + Some(Ok(frame)) + } + } + + fn block_on(f: F) -> F::Output { + futures::executor::block_on(f) + } + + #[test] + fn uart_reads_queued_chunks_then_reports_eof() { + let mut uart = EmbassyUart::new(MockRx(vec![b"hi".to_vec()]), MockTx::default()); + block_on(async { + let mut buf = [0u8; 8]; + assert_eq!(uart.read(&mut buf).await.unwrap(), 2); + assert_eq!(&buf[..2], b"hi"); + assert_eq!(uart.read(&mut buf).await.unwrap(), 0, "EOF is Ok(0)"); + }); + } + + #[test] + fn uart_writes_every_byte_and_flushes() { + let mut uart = EmbassyUart::new(MockRx(vec![]), MockTx::default()); + block_on(async { + uart.write_all(b"payload").await.unwrap(); + uart.flush().await.unwrap(); + }); + assert_eq!(uart.tx.written, b"payload"); + assert_eq!(uart.tx.flushes, 1); + } + + /// The UART drives core's framed connection, which is what the serial + /// connector rides. + #[test] + fn uart_drives_a_framed_connection() { + let rx = MockRx(vec![vec![2, b'h', b'i'], vec![3, b'y', b'e', b's']]); + let mut conn: FramedConnection<_, LenFramer, 64, 64> = FramedConnection::new( + EmbassyUart::new(rx, MockTx::default()), + LenFramer::default(), + ); + + block_on(async { + assert_eq!(conn.recv().await.unwrap(), Some(b"hi".to_vec())); + assert_eq!(conn.recv().await.unwrap(), Some(b"yes".to_vec())); + assert_eq!(conn.recv().await.unwrap(), None); + conn.send(b"ack").await.unwrap(); + }); + } + + /// The force-`Send` has to survive as far as the runner's boxed + /// `dyn Connection`. + #[test] + fn a_uart_connection_is_boxable_as_a_send_dyn_connection() { + let conn: FramedConnection<_, LenFramer, 64, 64> = FramedConnection::new( + EmbassyUart::new(MockRx(vec![]), MockTx::default()), + LenFramer::default(), + ); + let _boxed: alloc::boxed::Box = alloc::boxed::Box::new(conn); + } +} diff --git a/aimdb-embassy-adapter/src/lib.rs b/aimdb-embassy-adapter/src/lib.rs index 941d17a6..63cdc955 100644 --- a/aimdb-embassy-adapter/src/lib.rs +++ b/aimdb-embassy-adapter/src/lib.rs @@ -37,6 +37,11 @@ pub mod send_wrapper; #[cfg(all(not(feature = "std"), feature = "connectors"))] pub mod connectors; +// Embassy byte sources behind core's I/O traits that need no network stack — +// a UART is not a socket, so it does not pay for one. +#[cfg(all(not(feature = "std"), feature = "connector-io"))] +pub mod io; + // Embassy implementations of core's runtime-neutral I/O traits, so connector // crates stay runtime-neutral. #[cfg(all(not(feature = "std"), feature = "net"))] diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index ad8d3a1b..e947a34a 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -363,56 +363,6 @@ impl StreamListener for EmbassyTcpListener { } } -// =========================================================================== -// UART. -// =========================================================================== - -/// A UART, or any `embedded-io-async` read/write pair, as one [`ByteStream`] — -/// so the connector names no `embedded-io-async` types of its own. -pub struct EmbassyUart { - rx: Rd, - tx: Wr, -} - -// SAFETY: single-core cooperative Embassy executor — see the module invariant. -unsafe impl Send for EmbassyUart {} - -impl EmbassyUart { - /// Present an already-split UART's halves as one stream. - pub fn new(rx: Rd, tx: Wr) -> Self { - Self { rx, tx } - } -} - -impl ByteStream for EmbassyUart -where - Rd: embedded_io_async::Read, - Wr: embedded_io_async::Write, -{ - fn read<'a>( - &'a mut self, - buf: &'a mut [u8], - ) -> impl Future> + Send + 'a { - SendFutureWrapper(async move { self.rx.read(buf).await.map_err(|_| TransportError::Io) }) - } - - fn write_all<'a>( - &'a mut self, - buf: &'a [u8], - ) -> impl Future> + Send + 'a { - SendFutureWrapper(async move { - self.tx - .write_all(buf) - .await - .map_err(|_| TransportError::Closed) - }) - } - - fn flush(&mut self) -> impl Future> + Send + '_ { - SendFutureWrapper(async move { self.tx.flush().await.map_err(|_| TransportError::Closed) }) - } -} - // =========================================================================== // Datagrams. // =========================================================================== @@ -641,135 +591,8 @@ where #[cfg(test)] mod tests { - use super::*; - use aimdb_core::session::{Connection, FrameFault, FramedConnection, Framer}; - use alloc::vec; - use alloc::vec::Vec; - // `host_test_stubs!()` must expand once per binary; `buffer` already does - // it for this one. - - /// An in-memory read half: hands out queued chunks, then EOF. - struct MockRx(Vec>); - - impl embedded_io_async::ErrorType for MockRx { - type Error = embedded_io_async::ErrorKind; - } - - impl embedded_io_async::Read for MockRx { - async fn read(&mut self, buf: &mut [u8]) -> Result { - if self.0.is_empty() { - return Ok(0); - } - let chunk = self.0.remove(0); - let n = chunk.len().min(buf.len()); - buf[..n].copy_from_slice(&chunk[..n]); - Ok(n) - } - } - - /// An in-memory write half recording what it was given. - #[derive(Default)] - struct MockTx { - written: Vec, - flushes: usize, - } - - impl embedded_io_async::ErrorType for MockTx { - type Error = embedded_io_async::ErrorKind; - } - - impl embedded_io_async::Write for MockTx { - async fn write(&mut self, buf: &[u8]) -> Result { - self.written.extend_from_slice(buf); - Ok(buf.len()) - } - async fn flush(&mut self) -> Result<(), Self::Error> { - self.flushes += 1; - Ok(()) - } - } - - /// Length-prefixed framer, enough to drive a `FramedConnection`. - #[derive(Default)] - struct LenFramer { - buf: Vec, - } - - impl Framer for LenFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { - out.push(frame.len() as u8); - out.extend_from_slice(frame); - Ok(()) - } - fn push_bytes(&mut self, bytes: &[u8]) { - self.buf.extend_from_slice(bytes); - } - fn next_frame(&mut self) -> Option, FrameFault>> { - let len = *self.buf.first()? as usize; - if self.buf.len() < len + 1 { - return None; - } - let frame = self.buf[1..len + 1].to_vec(); - self.buf.drain(..len + 1); - Some(Ok(frame)) - } - } - - fn block_on(f: F) -> F::Output { - futures::executor::block_on(f) - } - - #[test] - fn uart_reads_queued_chunks_then_reports_eof() { - let mut uart = EmbassyUart::new(MockRx(vec![b"hi".to_vec()]), MockTx::default()); - block_on(async { - let mut buf = [0u8; 8]; - assert_eq!(uart.read(&mut buf).await.unwrap(), 2); - assert_eq!(&buf[..2], b"hi"); - assert_eq!(uart.read(&mut buf).await.unwrap(), 0, "EOF is Ok(0)"); - }); - } - - #[test] - fn uart_writes_every_byte_and_flushes() { - let mut uart = EmbassyUart::new(MockRx(vec![]), MockTx::default()); - block_on(async { - uart.write_all(b"payload").await.unwrap(); - uart.flush().await.unwrap(); - }); - assert_eq!(uart.tx.written, b"payload"); - assert_eq!(uart.tx.flushes, 1); - } - - /// The UART drives core's framed connection, which is what the serial - /// connector rides. - #[test] - fn uart_drives_a_framed_connection() { - let rx = MockRx(vec![vec![2, b'h', b'i'], vec![3, b'y', b'e', b's']]); - let mut conn: FramedConnection<_, LenFramer, 64, 64> = FramedConnection::new( - EmbassyUart::new(rx, MockTx::default()), - LenFramer::default(), - ); - - block_on(async { - assert_eq!(conn.recv().await.unwrap(), Some(b"hi".to_vec())); - assert_eq!(conn.recv().await.unwrap(), Some(b"yes".to_vec())); - assert_eq!(conn.recv().await.unwrap(), None); - conn.send(b"ack").await.unwrap(); - }); - } - - /// The force-`Send` has to survive as far as the runner's boxed - /// `dyn Connection`. - #[test] - fn a_uart_connection_is_boxable_as_a_send_dyn_connection() { - let conn: FramedConnection<_, LenFramer, 64, 64> = FramedConnection::new( - EmbassyUart::new(MockRx(vec![]), MockTx::default()), - LenFramer::default(), - ); - let _boxed: alloc::boxed::Box = alloc::boxed::Box::new(conn); - } + // it for this one. The UART's own tests moved with it to `crate::io`. /// The host time driver pins the clock at 0, so only an already-expired /// sleep can be driven here. @@ -777,6 +600,6 @@ mod tests { #[test] fn delay_completes_an_already_expired_sleep() { use aimdb_core::session::Delay; - block_on(EmbassyDelay.sleep(core::time::Duration::ZERO)); + futures::executor::block_on(super::EmbassyDelay.sleep(core::time::Duration::ZERO)); } } diff --git a/aimdb-embassy-adapter/tests/connectors_smoke.rs b/aimdb-embassy-adapter/tests/connectors_smoke.rs deleted file mode 100644 index f3ade9f1..00000000 --- a/aimdb-embassy-adapter/tests/connectors_smoke.rs +++ /dev/null @@ -1,224 +0,0 @@ -//! Host smoke for the centralized Embassy connector spine — the **server** side -//! the serial smoke test doesn't reach: [`OneShotCell`]'s take-once semantics, -//! [`OneShotListener`]'s park-forever second `accept`, and core's `serve` over -//! the one-shot listener — the exact path `EmbassySessionServer::build` (and the -//! serial `SerialServer`) drives on an MCU. -//! -//! Runs under the adapter's host test feature set (`alloc,…,connectors`); the -//! futures are driven by `futures::executor::block_on`, no executor needed. - -#![cfg(all(not(feature = "std"), feature = "connectors"))] - -use std::sync::{Arc, Mutex}; - -use core::future::Future; -use core::pin::Pin; -use core::task::{Context, Poll}; - -use futures::executor::block_on; -use futures::future::{select, Either}; -use futures::pin_mut; - -use aimdb_core::session::{ - serve, AuthError, BoxFut, CodecError, Connection, Dispatch, EnvelopeCodec, Inbound, Listener, - Outbound, Payload, PeerInfo, RpcError, Session, SessionConfig, SessionCtx, SessionLimits, - TransportResult, -}; -use aimdb_embassy_adapter::connectors::{OneShotCell, OneShotListener}; - -/// Yields (self-waking) `n` times, then completes — bounds how long we drive a -/// never-returning future like `serve` under `block_on`. -struct YieldN(usize); - -impl Future for YieldN { - type Output = (); - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { - if self.0 == 0 { - Poll::Ready(()) - } else { - self.0 -= 1; - cx.waker().wake_by_ref(); - Poll::Pending - } - } -} - -/// A scripted connection: `recv` yields the queued frames then `None` (EOF); -/// `send` records every frame into the shared log. -struct ScriptedConn { - /// Frames to yield, in reverse order (popped from the back). - inbox: Vec>, - sent: Arc>>>, - peer: PeerInfo, -} - -impl ScriptedConn { - fn new(inbox: Vec>, sent: Arc>>>) -> Self { - Self { - inbox, - sent, - peer: PeerInfo::default(), - } - } -} - -impl Connection for ScriptedConn { - fn recv(&mut self) -> BoxFut<'_, TransportResult>>> { - Box::pin(async move { Ok(self.inbox.pop()) }) - } - fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> { - self.sent.lock().unwrap().push(frame.to_vec()); - Box::pin(async { Ok(()) }) - } - fn peer(&self) -> &PeerInfo { - &self.peer - } -} - -/// Minimal server wire: a request frame is `[id:8][params]`; the reply echoes -/// the same shape back. Client-direction methods are unused here. -struct EchoCodec; - -impl EnvelopeCodec for EchoCodec { - fn decode(&self, frame: &[u8]) -> Result { - if frame.len() < 8 { - return Err(CodecError::Malformed); - } - Ok(Inbound::Request { - id: u64::from_be_bytes(frame[0..8].try_into().unwrap()), - method: "echo".to_string(), - params: Payload::from(&frame[8..]), - }) - } - fn encode(&self, msg: Outbound<'_>, out: &mut Vec) -> Result<(), CodecError> { - match msg { - Outbound::Reply { - id, - result: Ok(payload), - } => { - out.extend_from_slice(&id.to_be_bytes()); - out.extend_from_slice(&payload); - Ok(()) - } - _ => Err(CodecError::Malformed), - } - } - fn encode_inbound(&self, _msg: Inbound, _out: &mut Vec) -> Result<(), CodecError> { - Err(CodecError::Malformed) - } - fn decode_outbound<'a>(&self, _frame: &'a [u8]) -> Result, CodecError> { - Err(CodecError::Malformed) - } -} - -/// Accepts every peer; each session echoes a call's params back as the reply. -struct EchoDispatch; - -impl Dispatch for EchoDispatch { - fn authenticate<'a>( - &'a self, - _peer: &'a PeerInfo, - _first: Option<&'a [u8]>, - ) -> BoxFut<'a, Result> { - Box::pin(async { Ok(SessionCtx::default()) }) - } - fn open(&self, _ctx: &SessionCtx) -> Box { - Box::new(EchoSession) - } -} - -struct EchoSession; - -impl Session for EchoSession { - fn call<'a>( - &'a mut self, - _method: &'a str, - params: Payload, - ) -> BoxFut<'a, Result> { - Box::pin(async move { Ok(params) }) - } - fn write<'a>( - &'a mut self, - _topic: &'a str, - _payload: Payload, - ) -> BoxFut<'a, Result<(), RpcError>> { - Box::pin(async { Ok(()) }) - } -} - -fn session_config() -> SessionConfig { - SessionConfig { - limits: SessionLimits { - // The one-shot spine serves a single point-to-point peer. - max_connections: 1, - max_subs_per_connection: 4, - }, - reads_hello: false, - acks_subscribe: false, - } -} - -#[test] -fn one_shot_cell_hands_out_exactly_once() { - let cell = OneShotCell::new(42u32); - assert_eq!(cell.take(), Some(42)); - assert_eq!(cell.take(), None); - - let cell = OneShotCell::new("conn"); - assert_eq!(cell.take_required().unwrap(), "conn"); - // Second take: the canonical "already built" error, not a panic. - assert!(cell.take_required().is_err()); -} - -#[test] -fn one_shot_listener_parks_after_first_accept() { - let sent = Arc::new(Mutex::new(Vec::new())); - let mut listener = OneShotListener::new(ScriptedConn::new(vec![], sent)); - - block_on(async move { - listener - .accept() - .await - .expect("first accept hands out the connection"); - - // Point-to-point: the second accept must park forever, not error — an - // erroring accept would tear `serve` down. - let second = listener.accept(); - pin_mut!(second); - match select(second, YieldN(8)).await { - Either::Left(_) => panic!("second accept must park forever"), - Either::Right(((), _)) => {} - } - }); -} - -#[test] -fn serve_dispatches_over_one_shot_listener_then_keeps_waiting() { - let sent = Arc::new(Mutex::new(Vec::new())); - - // One scripted request (`id=7`, params `ping`), then EOF. - let mut request = 7u64.to_be_bytes().to_vec(); - request.extend_from_slice(b"ping"); - let conn = ScriptedConn::new(vec![request.clone()], sent.clone()); - - let serve_fut = serve( - OneShotListener::new(conn), - Arc::new(EchoCodec), - Arc::new(EchoDispatch), - session_config(), - ); - - block_on(async move { - pin_mut!(serve_fut); - // `serve` must process the session but never return: after the peer's - // EOF it loops back to the parked one-shot accept. - match select(serve_fut, YieldN(8)).await { - Either::Left(_) => panic!("serve must keep waiting on the parked accept"), - Either::Right(((), _)) => {} - } - }); - - // The echoed reply went out before EOF: same `[id:8][params]` bytes back. - let sent = sent.lock().unwrap(); - assert_eq!(sent.as_slice(), &[request]); -} diff --git a/aimdb-serial-connector/CHANGELOG.md b/aimdb-serial-connector/CHANGELOG.md index 8c8b0116..36712055 100644 --- a/aimdb-serial-connector/CHANGELOG.md +++ b/aimdb-serial-connector/CHANGELOG.md @@ -9,10 +9,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- **`tokio-runtime` now depends on `aimdb-tokio-adapter`** (feature `net`), so the - byte source is the adapter's on both runtimes rather than duplicated here. This - reverses the earlier decision to keep a concrete adapter off the public feature; - `_test-tokio` remains as an alias. +- **Features name what the code needs; no runtime appears in the public surface.** + `connector` gates the COBS framer and the sugar over it — `framed`, the + one-shot `Dialer`/`Listener`, `SerialClient`/`SerialServer` — which need core's + session layer and nothing else. Its library graph is `aimdb-core` + `cobs`, + with no adapter, so **Embassy is no longer a feature of this crate**: an + Embassy caller enables `connector` and passes `EmbassyUart`, the same line a + FreeRTOS caller writes with its own UART. +- **`std` gains a meaning it did not have.** It previously gated no code at all — + the sole `cfg(feature = "std")` was the `no_std` attribute — and carried an + unused `thiserror` dependency, now dropped. It adds core's `std` plus the + `tokio-serial` port backend behind `SerialPortDialer`, the one item here that + needs a specific async runtime, and the 23 crates (`serialport`, `nix`, + `libc`, …) that come with it. It also pulls `aimdb-tokio-adapter` (feature + `net`), so the byte source is the adapter's on both runtimes rather than + duplicated here — reversing the earlier decision to keep a concrete adapter off + the public feature, and making the internal `_test-tokio` feature redundant. It + is removed; `std` gates the host tests and `serial_demo` directly. A host + caller wanting only the neutral half enables `connector`, which builds fine on + std. +- **`EmbassyFramed` and `TokioFramed` are removed (breaking).** Both + were one-liners over the generic `SerialFramed`, neither had a call site + outside this crate's own tests, and naming an adapter in their definition was + the only thing forcing an adapter dependency into the features. Write + `SerialFramed>` / `SerialFramed>`. +- **`tokio-runtime` and `embassy-runtime` are gone**, with no aliases: consumers + move to `std` and `connector` respectively. The `thumbv7em` type-check that + keeps the two byte sources on one code path survives as the internal + `_test-embassy` feature. +- **One path for both runtimes (breaking).** `SerialServer::new` takes an + adapter byte stream (`EmbassyUart::new(rx, tx)`, `TokioByteStream(port)`) + instead of `(path, baud)` or split UART halves; the application opens the + device. `SerialClient::new(stream)` serves it once — with + `reconnect` off, since a moved-in stream cannot be reopened — and + `SerialClient::over_port(path, baud)` reopens and redials on a host. + `tokio_transport` and `embassy_transport` are deleted with the whole + `Tokio*`/`Embassy*` alias set — `SerialDialer` is now `SerialPortDialer`, and + `SerialListener`/`TokioSerialConnection` are gone. +- **The host read chunk drops 256 → 64.** The deleted `TokioSerialConnection` + carried its own `READ_CHUNK = 256`; both paths now share `framing::READ_CHUNK`, + sized for an MCU UART ring, so a host reading a real port issues four times the + `read` calls per kilobyte. Kept deliberately — one code path, one chunk size — + and `SerialFramed`'s chunks are const generics, so a host-specific value stays + available without structural change. - **Reports through the `log_*` facade instead of `tracing::` directly** (design 050 §10.5), so a `log` destination — an FFI layer's, say — sees this crate's events too. Each call site also shed the hand-written @@ -22,6 +61,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`connector` — runtime-neutral client and server sugar.** `SerialServer` + and `SerialClient` over any adapter byte stream, plus `OneShotDialer` / + `OneShotListener`: a UART is point-to-point, so the stream is served once — + the dialer then errors, the listener parks, because `serve` loops on `accept` + and would spin on an error. - **`framing` gains core's `Framer` — the connector reduced to framing.** `CobsFramer` against core's `Framer` plus core's `FramedConnection` serve both runtimes; the byte sources come from the adapters (`TokioByteStream`, `EmbassyUart`), so this crate diff --git a/aimdb-serial-connector/Cargo.toml b/aimdb-serial-connector/Cargo.toml index fa3f1375..64d59124 100644 --- a/aimdb-serial-connector/Cargo.toml +++ b/aimdb-serial-connector/Cargo.toml @@ -13,41 +13,42 @@ categories = ["embedded", "network-programming", "no-std"] [features] default = ["aimdb-core/alloc"] -# std build (host). Pulls the tokio half's std bits. -std = [ - "aimdb-core/std", +# The COBS framer and the runtime-neutral sugar over it: `framed`, the one-shot +# `Dialer`/`Listener`, `SerialClient`/`SerialServer`. Needs core's session layer +# and nothing else — no adapter, no `std` — so any runtime enables just this and +# hands `framed` its own `ByteStream`. That includes Embassy: the caller passes +# `EmbassyUart`, exactly as a FreeRTOS caller passes its own UART, which is why +# there is no runtime feature here to choose between them. +connector = [ "aimdb-core/alloc", "aimdb-core/connector-session", - "thiserror", + "aimdb-core/remote", ] -# Host / gateway half: real serial via `tokio-serial`, riding the generic -# `SessionClientConnector`/`SessionServerConnector` from core. -tokio-runtime = [ - "std", - "aimdb-core/connector-session", - "aimdb-core/remote", +# Host build. Adds core's `std` plus the `tokio-serial` port backend behind +# `SerialPortDialer` — the one item here that needs a specific async runtime, and +# 23 crates (`serialport`, `nix`, `libc`, …) that a neutral build does not pay +# for. A host caller wanting only the neutral half enables `connector` instead; +# it builds fine on std. +std = [ + "connector", + "aimdb-core/std", "dep:tokio", "dep:tokio-serial", - # The adapter owns the byte source, as it does on Embassy; this crate - # contributes only the COBS framer. "dep:aimdb-tokio-adapter", "aimdb-tokio-adapter/net", ] -# Embedded half: `no_std + alloc`, generic over `embedded-io-async` UART halves. -# Thin sugar over the centralized Embassy session spine in `aimdb-embassy-adapter` -# (`connector-io`), which owns the single-core `unsafe` + force-`Send`; this crate -# contributes only the COBS `Framer`. -embassy-runtime = [ - "aimdb-core/alloc", - "aimdb-core/connector-session", - "aimdb-core/remote", +# Internal: type-checks the Embassy byte source against the same `SerialFramed` +# the host uses, on `thumbv7em`, so a divergence between the two paths fails here +# rather than in a demo. Not part of the public surface — an Embassy consumer +# enables `connector` and brings its own adapter. +_test-embassy = [ + "connector", "dep:aimdb-embassy-adapter", + # `connector-io` carries `EmbassyUart` and the framed spine; no `net`, so + # this leg type-checks a UART without compiling a TCP/IP stack. "aimdb-embassy-adapter/connector-io", - # `EmbassyUart`, the neutral byte source the COBS framer rides. - "aimdb-embassy-adapter/net", - # `embassy-time`/`-sync` give the `EmbassyAdapter` clock the smoke test runs on. "aimdb-embassy-adapter/embassy-time", "aimdb-embassy-adapter/embassy-sync", "dep:embedded-io-async", @@ -61,10 +62,6 @@ tracing = ["aimdb-core/tracing"] log = ["aimdb-core/log"] defmt = ["dep:defmt", "aimdb-core/defmt"] -# Retained as an alias: `tokio-runtime` now carries the adapter itself, but the -# host integration test and `serial_demo` still gate on this name. Kept out of -# `[dev-dependencies]` so the no_std `embassy-runtime` test build never sees it. -_test-tokio = ["tokio-runtime"] [dependencies] # AimX protocol (codec + dispatch) and the generic session connectors live in @@ -76,13 +73,12 @@ cobs = { workspace = true, features = ["alloc"] } # --- tokio half (std) --- tokio = { workspace = true, optional = true, features = ["io-util"] } tokio-serial = { workspace = true, optional = true } -thiserror = { workspace = true, optional = true } # --- embassy half (no_std) --- aimdb-embassy-adapter = { version = "0.6.0", path = "../aimdb-embassy-adapter", default-features = false, optional = true } embedded-io-async = { workspace = true, optional = true } -# --- test-only (see the `_test-tokio` feature) --- +# --- test-only --- aimdb-tokio-adapter = { version = "0.6.0", path = "../aimdb-tokio-adapter", optional = true } # --- logging --- @@ -93,7 +89,7 @@ defmt = { workspace = true, optional = true } # (no-op defmt logger + host time driver, 035 §2.4); the expansion references # `defmt` and `embassy-time-driver` at the invocation site, so both must stay # resolvable in the test binary. (`aimdb-embassy-adapter` itself is the regular -# `embassy-runtime` optional dep — keeping it out of dev-deps avoids a +# `_test-embassy` optional dep — keeping it out of dev-deps avoids a # std/no_std unification clash with the tokio tests, where it would otherwise # compile against a std `aimdb-core`.) embassy-time-driver = "0.2.2" @@ -108,8 +104,7 @@ futures = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -# The real-serial host demo needs a concrete adapter (and tokio-serial); gate it on -# the internal test feature so it doesn't pull an adapter into the embassy build. +# The real-serial host demo needs the port backend, which only `std` supplies. [[example]] name = "serial_demo" -required-features = ["_test-tokio"] +required-features = ["std"] diff --git a/aimdb-serial-connector/examples/serial_demo.rs b/aimdb-serial-connector/examples/serial_demo.rs index 8d1a7a6d..e4de99a6 100644 --- a/aimdb-serial-connector/examples/serial_demo.rs +++ b/aimdb-serial-connector/examples/serial_demo.rs @@ -6,7 +6,7 @@ //! //! ```text //! # board (Embassy SerialServer) ⇄ host: -//! cargo run --example serial_demo --features _test-tokio -- client /dev/ttyACM0 +//! cargo run --example serial_demo --features std -- client /dev/ttyACM0 //! //! # host SerialServer ⇄ host client over a PTY (no hardware): //! socat -d -d pty,raw,echo=0 pty,raw,echo=0 # prints two /dev/pts/N @@ -22,8 +22,8 @@ //! - `raw [baud] [method] [name]` — low-level debug: send one request and //! print the full decoded reply (no engine), handy when `client` misbehaves. //! -//! Built only under the internal `_test-tokio` feature (it needs a concrete -//! adapter); see the crate's `Cargo.toml`. +//! Built only under `std` (it needs the `tokio-serial` port backend); see the +//! crate's `Cargo.toml`. //! //! On macOS the board's VCP is `/dev/cu.usbmodem…` (use the `cu.*`, not `tty.*`, //! node). Run from the workspace root, and make sure nothing else holds the port @@ -37,7 +37,8 @@ use aimdb_core::remote::AimxConfig; use aimdb_core::session::aimx::AimxCodec; use aimdb_core::session::{run_client, ClientConfig, Payload}; use aimdb_core::AimDbBuilder; -use aimdb_serial_connector::tokio_transport::{SerialDialer, SerialServer}; +use aimdb_serial_connector::connector::{SerialPortDialer, SerialServer}; +use aimdb_tokio_adapter::net::TokioByteStream; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use serde::{Deserialize, Serialize}; use serde_json::json; @@ -172,7 +173,7 @@ async fn run_set_mode(device: String, baud: u32) { println!("[set] writing `setting` over {device} @ {baud} baud"); let (handle, engine) = run_client( - SerialDialer::new(device, baud), + SerialPortDialer::new(device, baud), AimxCodec, ClientConfig { sends_hello: false, @@ -217,9 +218,13 @@ async fn run_set_mode(device: String, baud: u32) { async fn run_server(device: String, baud: u32) { println!("[server] serving AimX over {device} @ {baud} baud (Ctrl-C to stop)"); + // The application opens the device; the connector only frames it. + let port = open_port(&device, baud); let mut builder = AimDbBuilder::new() .runtime(Arc::new(TokioAdapter)) - .with_connector(SerialServer::new(device, baud).with_config(AimxConfig::uds_default())); + .with_connector( + SerialServer::new(TokioByteStream(port)).with_config(AimxConfig::uds_default()), + ); builder.configure::("counter", |reg| { reg.buffer(BufferCfg::SingleLatest).with_remote_access(); }); @@ -246,7 +251,7 @@ async fn run_client_mode(device: String, baud: u32) { println!("[client] querying AimX over {device} @ {baud} baud"); let (handle, engine) = run_client( - SerialDialer::new(device, baud), + SerialPortDialer::new(device, baud), AimxCodec, ClientConfig { sends_hello: false, @@ -272,3 +277,15 @@ async fn run_client_mode(device: String, baud: u32) { tokio::time::sleep(Duration::from_secs(1)).await; } } + +/// Open a serial device, or exit with a readable message. +fn open_port(device: &str, baud: u32) -> tokio_serial::SerialStream { + use tokio_serial::SerialPortBuilderExt; + match tokio_serial::new(device, baud).open_native_async() { + Ok(port) => port, + Err(e) => { + eprintln!("[server] cannot open {device} @ {baud} baud: {e}"); + std::process::exit(1); + } + } +} diff --git a/aimdb-serial-connector/src/connector.rs b/aimdb-serial-connector/src/connector.rs new file mode 100644 index 00000000..56b80776 --- /dev/null +++ b/aimdb-serial-connector/src/connector.rs @@ -0,0 +1,233 @@ +//! Runtime-neutral serial client and server sugar. +//! +//! Both take a byte stream from an adapter — `EmbassyUart` on the MCU, +//! `TokioByteStream` over a `SerialStream` on the host — and this crate +//! contributes only the COBS [`CobsFramer`]. +//! +//! A UART is point-to-point, so the stream is moved in and served once; there +//! is no accept loop. `SerialPortDialer` (feature `std`) is the one exception, +//! because a host can reopen a device by path. + +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 aimdb_core::connector::ConnectorBuilder; +use aimdb_core::log_info; +use aimdb_core::remote::{AimxConfig, SecurityPolicy}; +use aimdb_core::session::aimx::{AimxCodec, AimxDispatch}; +use aimdb_core::session::{ + ByteStream, ClientConfig, Dispatch, FramedConnection, OneShot, OneShotDialer, OneShotListener, + SessionClientConnector, SessionConfig, SessionLimits, SessionServerConnector, +}; +// `SerialPortDialer` is the only item here that dials anything itself, and it is +// `std`-only; everything else rides core's one-shots. +#[cfg(feature = "std")] +use aimdb_core::session::{BoxFut, Connection, Dialer, TransportError, TransportResult}; +use aimdb_core::{AimDb, DbError, DbResult}; + +use crate::framing::{CobsFramer, READ_CHUNK, WRITE_CHUNK}; +use crate::DEFAULT_SCHEME; + +type BoxFuture = Pin + Send + 'static>>; +type BuildFuture<'a> = Pin>> + Send + 'a>>; + +/// A COBS-framed connection over any adapter byte stream. +pub type SerialFramed = FramedConnection; + +/// Frame an adapter's byte stream with COBS. +pub fn framed(stream: S) -> SerialFramed { + FramedConnection::new(stream, CobsFramer::new()) +} + +/// Opens a serial device by path on each [`connect`](Dialer::connect). +/// +/// The one redialable serial dialer: a host can reopen a device, so +/// `run_client` reconnects after a drop. Cheap to clone (path plus baud). +/// `tokio-serial` stays here rather than in the adapter — opening a tty is not +/// a runtime concern. +#[cfg(feature = "std")] +#[derive(Clone)] +pub struct SerialPortDialer { + path: String, + baud: u32, +} + +#[cfg(feature = "std")] +impl SerialPortDialer { + /// Dial the serial device at `path` (e.g. `/dev/ttyUSB0`) at `baud`. + pub fn new(path: impl Into, baud: u32) -> Self { + Self { + path: path.into(), + baud, + } + } +} + +#[cfg(feature = "std")] +impl Dialer for SerialPortDialer { + fn connect(&self) -> BoxFut<'_, TransportResult>> { + Box::pin(async move { + use tokio_serial::SerialPortBuilderExt; + let stream = tokio_serial::new(&self.path, self.baud) + .open_native_async() + .map_err(|_| TransportError::Io)?; + // Discard bytes a previous session left in the OS input buffer; + // otherwise the first frame is a stale leftover that fails to decode + // and desyncs the stream until the next COBS sentinel. + use tokio_serial::SerialPort; + let _ = stream.clear(tokio_serial::ClearBuffer::Input); + Ok( + Box::new(framed(aimdb_tokio_adapter::net::TokioByteStream(stream))) + as Box, + ) + }) + } +} + +/// Constructs a serial session client connector. +pub struct SerialClient; + +impl SerialClient { + /// Mirror records to and from the AimX peer on `stream`, served once. + /// + /// Reconnect is **disabled** (unlike `ClientConfig::default`): the stream is + /// moved in and cannot be re-acquired. `run_client` would stop anyway — it + /// treats a dialer's `TransportError::Closed` as terminal — but saying so + /// here keeps the intent local rather than resting on that two-crate + /// handshake, and skips a pointless backoff and "dial failed" warning on the + /// way out. A caller whose stream really can be redialed opts back in with + /// `.with_config(...)`; on a host, prefer `over_port`, which reopens the + /// device for real. + /// + /// (Both names are unlinked deliberately: `TransportError` and `over_port` + /// only exist under `std`, and this item does not.) + #[allow(clippy::new_ret_no_self)] + pub fn new(stream: S) -> SessionClientConnector>, AimxCodec> + where + S: ByteStream + Send + 'static, + { + SessionClientConnector::new(OneShotDialer::new(framed(stream)), AimxCodec) + .scheme(DEFAULT_SCHEME) + .with_config(ClientConfig { + reconnect: false, + ..ClientConfig::default() + }) + } + + /// Mirror records over a serial device this process opens by path, + /// reconnecting after a drop. + /// + /// Keeps `ClientConfig`'s default `reconnect: true` — unlike + /// [`new`](Self::new), this dialer can genuinely reopen the device. + #[cfg(feature = "std")] + pub fn over_port( + path: impl Into, + baud: u32, + ) -> SessionClientConnector { + SessionClientConnector::new(SerialPortDialer::new(path, baud), AimxCodec) + .scheme(DEFAULT_SCHEME) + } +} + +/// Serves AimX over a moved-in serial stream. +pub struct SerialServer { + stream: OneShot, + config: AimxConfig, + scheme: String, +} + +impl SerialServer { + /// Serve AimX over `stream`. + pub fn new(stream: S) -> Self { + Self { + stream: OneShot::new(stream), + config: AimxConfig::uds_default(), + scheme: DEFAULT_SCHEME.to_string(), + } + } + + /// Use a prepared [`AimxConfig`] for limits and security policy. + pub fn with_config(mut self, config: AimxConfig) -> Self { + self.config = config; + self + } + + /// Set the security policy. + pub fn security_policy(mut self, policy: SecurityPolicy) -> Self { + self.config = self.config.security_policy(policy); + self + } + + /// Maximum live subscriptions per connection. + pub fn max_subs_per_connection(mut self, max: usize) -> Self { + self.config = self.config.max_subs_per_connection(max); + self + } + + /// Override the scheme this connector registers. + pub fn scheme(mut self, scheme: impl Into) -> Self { + self.scheme = scheme.into(); + self + } +} + +impl ConnectorBuilder for SerialServer +where + S: ByteStream + Send + 'static, +{ + fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { + let config = self.config.clone(); + let scheme = self.scheme.clone(); + Box::pin(async move { + // Taken on first poll, not at call time: a `build()` future dropped + // before it is polled must leave the stream where it was. + let stream = self + .stream + .take() + .ok_or_else(|| DbError::InvalidOperation { + operation: "SerialServer::build".to_string(), + reason: "the moved-in stream was already taken; build() ran twice".to_string(), + })?; + log_info!("Initializing AimX serial server on scheme '{}'", scheme); + let session_config = SessionConfig { + limits: SessionLimits { + // A UART carries a single peer. + max_connections: 1, + max_subs_per_connection: config.max_subs_per_connection, + }, + reads_hello: false, + // AimX's subscribe ack stays implicit (events flow); no ack frame. + acks_subscribe: false, + }; + let listener = OneShot::new(OneShotListener::new(framed(stream))); + let dispatch_config = config; + let connector = SessionServerConnector::new( + move || { + listener.take().ok_or_else(|| DbError::InvalidOperation { + operation: "SerialServer::build".to_string(), + reason: "the moved-in stream was already taken".to_string(), + }) + }, + AimxCodec, + move |db: &AimDb| -> Arc { + crate::apply_writable(db, &dispatch_config); + Arc::new(AimxDispatch::new( + Arc::new(db.clone()), + dispatch_config.clone(), + )) + }, + session_config, + ) + .scheme(scheme); + connector.build(db).await + }) + } + + fn scheme(&self) -> &str { + &self.scheme + } +} diff --git a/aimdb-serial-connector/src/embassy_transport.rs b/aimdb-serial-connector/src/embassy_transport.rs deleted file mode 100644 index 96ee9035..00000000 --- a/aimdb-serial-connector/src/embassy_transport.rs +++ /dev/null @@ -1,237 +0,0 @@ -//! Embassy serial transport (feature `embassy-runtime`, `no_std + alloc`) — thin -//! sugar over the centralized Embassy session spine in `aimdb-embassy-adapter`. -//! -//! This half contributes **only** the COBS [`Framer`] plus thin sugar; the framed -//! [`Connection`](aimdb_core::session::Connection), the one-shot -//! dialer/listener/cell, and the force-`Send` plumbing all live in -//! [`aimdb_embassy_adapter::connectors`]. So this module carries **no `unsafe`** -//! (down from the seven `unsafe impl`s this half used to hand-roll) — the Embassy -//! half is now structurally a sibling of the Tokio half, -//! both thin sugar over a shared spine. -//! -//! Generic over the `embedded-io-async` `Read`/`Write` halves (the common Embassy -//! HAL shape, e.g. `Uart::split()`), so it works with any chip's async UART. - -use core::future::Future; -use core::pin::Pin; - -use alloc::boxed::Box; -use alloc::string::{String, ToString}; -use alloc::sync::Arc; -use alloc::vec; -use alloc::vec::Vec; - -use embedded_io_async::{Read, Write}; - -use aimdb_embassy_adapter::connectors::{ - EmbassyConnection, EmbassySessionClient, Framer, OneShotCell, OneShotDialer, OneShotListener, -}; - -use aimdb_core::connector::ConnectorBuilder; -use aimdb_core::remote::{AimxConfig, SecurityPolicy}; -use aimdb_core::session::aimx::{AimxCodec, AimxDispatch}; -use aimdb_core::session::{serve, Dispatch, SessionConfig, SessionLimits}; -use aimdb_core::{AimDb, DbResult}; - -use crate::framing::{encode_frame, FrameAccumulator}; -use crate::DEFAULT_SCHEME; - -type BoxFuture = Pin + Send + 'static>>; -type BuildFuture<'a> = Pin>> + Send + 'a>>; - -/// How many bytes a single UART `read` pulls before re-checking for a frame. -const READ_CHUNK: usize = 64; -/// Max bytes per `write` call. Some HAL `BufferedUart::write` is atomic-or-error -/// (e.g. `embassy-stm32` returns `BufferTooLong` for a single write larger than -/// its TX ring), so a frame bigger than the buffer must be split. -const WRITE_CHUNK: usize = 64; - -/// The framed connection type a serial peripheral produces: COBS over the UART. -type SerialConnection = EmbassyConnection; - -// =========================================================================== -// COBS framer — the only serial-specific transport bit. -// =========================================================================== - -/// COBS framing for the Embassy [`EmbassyConnection`]: `encode` COBS-encodes a -/// frame and appends the `0x00` sentinel; the accumulator yields one frame per -/// sentinel (a malformed run is skipped — COBS is self-synchronizing). -pub struct CobsFramer { - acc: FrameAccumulator, -} - -impl CobsFramer { - /// A fresh COBS framer. - pub fn new() -> Self { - Self { - acc: FrameAccumulator::new(), - } - } -} - -impl Default for CobsFramer { - fn default() -> Self { - Self::new() - } -} - -impl Framer for CobsFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) { - encode_frame(frame, out); - } - - fn push_bytes(&mut self, bytes: &[u8]) { - self.acc.push_bytes(bytes); - } - - fn next_frame(&mut self) -> Option, ()>> { - // The accumulator's `FrameError` collapses to `()`: the connection only - // distinguishes "got a frame" from "skip and resync". - self.acc.next_frame().map(|r| r.map_err(|_| ())) - } -} - -// =========================================================================== -// Client sugar — one-shot dial over the moved-in UART. -// =========================================================================== - -/// Constructs an [`EmbassySessionClient`] that mirrors records to/from an AimX -/// peer over a serial UART. `SerialClient::new(rx, tx)` is sugar; chain -/// `.scheme(...)` / `.with_config(...)` on the returned connector and register it -/// with `with_connector`. -/// -/// Reconnect is disabled by default: the peripheral is moved in and can't be -/// re-acquired after a drop. -pub struct SerialClient; - -impl SerialClient { - /// Mirror records to/from the AimX peer over the split UART halves (e.g. from - /// `Uart::split()`). Scheme defaults to [`DEFAULT_SCHEME`]. - // Sugar constructor: intentionally returns the spine connector, not `Self`. - #[allow(clippy::new_ret_no_self)] - pub fn new( - rx: Rd, - tx: Wr, - ) -> EmbassySessionClient>, AimxCodec> - where - Rd: Read + 'static, - Wr: Write + 'static, - { - let conn = EmbassyConnection::new(rx, tx, CobsFramer::new()); - // Reconnect stays disabled (the spine's default): the UART peripheral is - // moved in and can't be re-acquired. - EmbassySessionClient::new(OneShotDialer::new(conn), AimxCodec).scheme(DEFAULT_SCHEME) - } -} - -// =========================================================================== -// Server sugar — serve the full AimX toolset over the moved-in UART. -// =========================================================================== - -/// Serves the full AimX toolset over a serial UART, so a host (or another board) -/// can `record.list`/`get`/`set`/`subscribe`/`drain` this db over the wire. -/// Register it directly with `with_connector` (illustrative — the UART halves -/// come from device init on a thumb target): -/// -/// ```ignore -/// builder.with_connector( -/// SerialServer::new(rx, tx).security_policy(SecurityPolicy::read_only()), -/// ); -/// ``` -/// -/// Holds the moved-in framed UART connection (built up front from the halves) in -/// the adapter's force-`Send + Sync` [`OneShotCell`]; `build` takes it, hands it -/// to a [`OneShotListener`], and drives `serve`. Storing it in the cell (rather -/// than a bare `RefCell`) keeps **all** the `unsafe` in the adapter — this crate -/// has none. -pub struct SerialServer { - conn: OneShotCell>, - config: AimxConfig, - scheme: String, -} - -impl SerialServer -where - Rd: Read + 'static, - Wr: Write + 'static, -{ - /// Serve AimX over the split UART halves, with the default read-only policy. - pub fn new(rx: Rd, tx: Wr) -> Self { - Self { - conn: OneShotCell::new(EmbassyConnection::new(rx, tx, CobsFramer::new())), - config: AimxConfig::uds_default(), - scheme: DEFAULT_SCHEME.to_string(), - } - } -} - -impl SerialServer { - /// Use a prepared [`AimxConfig`] for the security policy / limits (the - /// `socket_path` / `socket_permissions` fields are unused over serial). - pub fn with_config(mut self, config: AimxConfig) -> Self { - self.config = config; - self - } - - /// Set the security policy (read-only vs. per-record-writable). - pub fn security_policy(mut self, policy: SecurityPolicy) -> Self { - self.config = self.config.security_policy(policy); - self - } - - /// Maximum live subscriptions for the connection. - pub fn max_subs_per_connection(mut self, max: usize) -> Self { - self.config = self.config.max_subs_per_connection(max); - self - } - - /// Override the scheme this connector registers. - pub fn scheme(mut self, scheme: impl Into) -> Self { - self.scheme = scheme.into(); - self - } -} - -impl ConnectorBuilder for SerialServer -where - Rd: Read + 'static, - Wr: Write + 'static, -{ - fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { - // Take the moved-in connection out of `&self` (build runs once); the - // canonical "already built" error lives on the adapter's cell. - let conn = self.conn.take_required(); - let config = self.config.clone(); - Box::pin(async move { - let conn = conn?; - // Apply the security policy's writable marking so `record.list` reports - // the `writable` flag (the dispatch also enforces it). - crate::apply_writable(db, &config); - let session_config = SessionConfig { - limits: SessionLimits { - // A UART carries a single peer. - max_connections: 1, - max_subs_per_connection: config.max_subs_per_connection, - }, - reads_hello: false, - // AimX's subscribe ack stays implicit (events flow); no ack frame. - acks_subscribe: false, - }; - let dispatch: Arc = - Arc::new(AimxDispatch::new(Arc::new(db.clone()), config)); - // `serve` is `Send` here: the one-shot listener + framed connection - // force-`Send` their futures inside the adapter. - let fut: BoxFuture = Box::pin(serve( - OneShotListener::new(conn), - Arc::new(AimxCodec), - dispatch, - session_config, - )); - Ok(vec![fut]) - }) - } - - fn scheme(&self) -> &str { - &self.scheme - } -} diff --git a/aimdb-serial-connector/src/framing.rs b/aimdb-serial-connector/src/framing.rs index d28ddeff..a82c820b 100644 --- a/aimdb-serial-connector/src/framing.rs +++ b/aimdb-serial-connector/src/framing.rs @@ -7,22 +7,20 @@ //! joins mid-stream resynchronizes on the next sentinel. AimX JSON never contains //! a raw `0x00`, so the encoding is overhead-minimal (one byte per ~254). //! -//! This module is shared by both runtime halves and is pure `no_std + alloc`, so -//! the round-trip is unit-tested on the host without any transport. +//! This module is pure `no_std + alloc` and serves every runtime, so the +//! round-trip is unit-tested on the host without any transport. //! //! Two layers live here. [`encode_frame`] and [`FrameAccumulator`] are the COBS //! codec itself, with no dependency on the session substrate. `CobsFramer` -//! below — unlinked, as it exists only behind a runtime feature — is that codec -//! behind core's `Framer` trait, plus the -//! `FramedConnection` aliases it forms with each adapter's byte source — the -//! whole of what this crate contributes to a session, since the byte sources -//! come from the adapters and this crate names no socket or UART type of its -//! own. That half needs `aimdb_core::session`, so it is gated on the runtime -//! features that enable core's `connector-session`. +//! below — unlinked, as it exists only behind `connector` — is that codec behind +//! core's `Framer` trait: the whole of what this crate contributes to a session, +//! since the byte sources come from the adapters and this crate names no socket +//! or UART type of its own. That half needs `aimdb_core::session`, so it is +//! gated on `connector`, which enables core's `connector-session`. // Gated with the items that use it: the accumulator below is `alloc`-only and // builds without core's session layer. -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] use aimdb_core::session::FrameFault; use alloc::vec::Vec; @@ -160,11 +158,11 @@ impl FrameAccumulator { // =========================================================================== /// Per-`read` chunk, matching the UART ring size. -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] pub const READ_CHUNK: usize = 64; /// Per-`write_all` chunk: some HAL `BufferedUart::write` rejects a single write /// larger than its TX ring. -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] pub const WRITE_CHUNK: usize = 64; /// COBS framing against core's [`Framer`](aimdb_core::session::Framer), so one @@ -173,13 +171,13 @@ pub const WRITE_CHUNK: usize = 64; /// `encode` COBS-encodes a frame and appends the [`DELIM`] sentinel; the /// accumulator yields one frame per sentinel, skipping a malformed run (COBS is /// self-synchronizing). -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] #[derive(Default)] pub struct CobsFramer { acc: FrameAccumulator, } -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] impl CobsFramer { /// A fresh COBS framer. pub fn new() -> Self { @@ -187,7 +185,7 @@ impl CobsFramer { } } -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] impl aimdb_core::session::Framer for CobsFramer { fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { encode_frame(frame, out); @@ -207,32 +205,13 @@ impl aimdb_core::session::Framer for CobsFramer { } } -/// A framed connection over the Embassy adapter's UART halves. -#[cfg(feature = "embassy-runtime")] -pub type EmbassyFramed = aimdb_core::session::FramedConnection< - aimdb_embassy_adapter::net::EmbassyUart, - CobsFramer, - READ_CHUNK, - WRITE_CHUNK, ->; - -/// A framed connection over any Tokio byte source — a `tokio_serial::SerialStream` -/// in production, a `tokio::io::duplex()` pipe in tests. -#[cfg(feature = "tokio-runtime")] -pub type TokioFramed = aimdb_core::session::FramedConnection< - aimdb_tokio_adapter::net::TokioByteStream, - CobsFramer, - READ_CHUNK, - WRITE_CHUNK, ->; - /// The same framer and the same core connection over the Embassy UART, boxed as /// the runner takes it. /// /// Type-checking this on `thumbv7em` is what "one connector module, no runtime /// `cfg` on the code path" means concretely: if the two paths diverge, the /// embedded check fails here rather than in an example. -#[cfg(feature = "embassy-runtime")] +#[cfg(feature = "_test-embassy")] #[allow(dead_code)] fn _same_framed_connection_serves_the_uart(rx: Rd, tx: Wr) where @@ -240,10 +219,10 @@ where Wr: embedded_io_async::Write + Send + 'static, { use aimdb_core::session::Connection; - use aimdb_embassy_adapter::net::EmbassyUart; + use aimdb_embassy_adapter::io::EmbassyUart; use alloc::boxed::Box; - let conn: EmbassyFramed = - EmbassyFramed::new(EmbassyUart::new(rx, tx), CobsFramer::new()); + let conn: crate::connector::SerialFramed> = + crate::connector::framed(EmbassyUart::new(rx, tx)); let _boxed: Box = Box::new(conn); } diff --git a/aimdb-serial-connector/src/lib.rs b/aimdb-serial-connector/src/lib.rs index 4dc517b8..89860027 100644 --- a/aimdb-serial-connector/src/lib.rs +++ b/aimdb-serial-connector/src/lib.rs @@ -2,29 +2,25 @@ //! remote access over a serial line. //! //! A thin, swappable transport crate (the serial sibling of `aimdb-uds-connector`): -//! it contributes only the `Dialer`/`Listener`/`Connection` triple plus thin -//! sugar; the AimX codec (`AimxCodec`), dispatch (`AimxDispatch`), and the -//! runtime-neutral session engines are reused verbatim from `aimdb-core`. +//! it contributes only the COBS framing plus thin sugar; the AimX codec +//! (`AimxCodec`), dispatch (`AimxDispatch`), `Connection` itself +//! (`FramedConnection`), and the runtime-neutral session engines are reused +//! verbatim from `aimdb-core`. //! //! Core's session items are named unlinked throughout these docs: they exist -//! only when a runtime feature pulls in `aimdb-core/connector-session`, and a -//! link to them fails `cargo doc` on a build without one. +//! only when the `connector` feature pulls in `aimdb-core/connector-session`, +//! and a link to them fails `cargo doc` on a build without it. The same goes for +//! anything gated on `std`. //! //! The wire is the same compact AimX JSON as UDS, but framed with **COBS** //! (Consistent Overhead Byte Stuffing) and a `0x00` delimiter instead of a //! newline — self-synchronizing on a lossy/unframed serial medium. See //! [`framing`]. //! -//! # Two halves -//! -//! - **`tokio-runtime`** (std, host/gateway): real serial via `tokio-serial`, -//! riding the generic `SessionClientConnector` / `SessionServerConnector`. -//! See `tokio_transport`. -//! - **`embassy-runtime`** (`no_std + alloc`, MCU): generic over -//! `embedded-io-async` UART halves; the COBS `Framer` plus thin sugar over the -//! centralized Embassy session spine in `aimdb-embassy-adapter`, which owns the -//! force-`Send` plumbing, the framed connection, and all the `unsafe` — this -//! crate carries none. See `embassy_transport`. +//! One path for both runtimes: the byte source comes from an adapter +//! (`EmbassyUart` on the MCU, `TokioByteStream` over a `SerialStream` on the +//! host) and this crate contributes only the COBS framer. A UART is +//! point-to-point, so the stream is moved in and served once. //! //! Both speak the `serial://` scheme by default ([`DEFAULT_SCHEME`]). @@ -32,17 +28,12 @@ extern crate alloc; -// The COBS codec, and (under either runtime feature) that codec behind core's -// `Framer` plus the `FramedConnection` aliases it forms with each adapter's -// byte source. Supersedes the two per-runtime transport modules below, which it -// will replace outright. +// The COBS codec, and — under `connector` — that codec behind core's `Framer`. pub mod framing; -#[cfg(feature = "tokio-runtime")] -pub mod tokio_transport; - -#[cfg(feature = "embassy-runtime")] -pub mod embassy_transport; +// Runtime-neutral `SerialClient`/`SerialServer` over an adapter's byte stream. +#[cfg(feature = "connector")] +pub mod connector; /// The default scheme `SerialClient`/`SerialServer` register when none is given. /// @@ -54,7 +45,7 @@ pub const DEFAULT_SCHEME: &str = "serial"; /// Mark each record named in the policy's writable set as writable, so /// `record.list` advertises the `writable` flag (the dispatch also enforces it). /// Shared by both `SerialServer` halves; mirrors the UDS connector. -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] pub(crate) fn apply_writable(db: &aimdb_core::AimDb, config: &aimdb_core::remote::AimxConfig) { for key in config.security_policy.writable_records() { if let Some(id) = db.inner().resolve_str(&key) { @@ -65,19 +56,8 @@ pub(crate) fn apply_writable(db: &aimdb_core::AimDb, config: &aimdb_core::remote } } -// Prefer the tokio names when both halves are compiled (e.g. host tests). -#[cfg(all(feature = "tokio-runtime", not(feature = "embassy-runtime")))] -pub use tokio_transport::{SerialClient, SerialDialer, SerialListener, SerialServer}; - -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use embassy_transport::{ - SerialClient as EmbassySerialClient, SerialServer as EmbassySerialServer, -}; -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use tokio_transport::{ - SerialClient as TokioSerialClient, SerialDialer, SerialListener, - SerialServer as TokioSerialServer, -}; +#[cfg(feature = "connector")] +pub use connector::{framed, SerialClient, SerialServer}; -#[cfg(all(feature = "embassy-runtime", not(feature = "tokio-runtime")))] -pub use embassy_transport::{SerialClient, SerialServer}; +#[cfg(feature = "std")] +pub use connector::SerialPortDialer; diff --git a/aimdb-serial-connector/src/tokio_transport.rs b/aimdb-serial-connector/src/tokio_transport.rs deleted file mode 100644 index 5beffe36..00000000 --- a/aimdb-serial-connector/src/tokio_transport.rs +++ /dev/null @@ -1,312 +0,0 @@ -//! tokio serial transport (feature `tokio-runtime`) — a [`Connection`] over an -//! async byte stream with COBS framing in the transport, plus -//! [`SerialClient`]/[`SerialServer`] sugar over the generic core connectors. -//! -//! The connection is generic over `AsyncRead + AsyncWrite` so it backs a real -//! `tokio_serial::SerialStream` in production and a `tokio::io::duplex()` pipe in -//! tests — no hardware needed. - -use std::future::Future; -use std::pin::Pin; -use std::sync::Arc; - -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio_serial::{SerialPortBuilderExt, SerialStream}; - -use aimdb_core::connector::ConnectorBuilder; -use aimdb_core::log_info; -use aimdb_core::remote::{AimxConfig, SecurityPolicy}; -use aimdb_core::session::aimx::{AimxCodec, AimxDispatch}; -use aimdb_core::session::{ - BoxFut, Connection, Dialer, Dispatch, Listener, PeerInfo, SessionClientConnector, - SessionConfig, SessionLimits, SessionServerConnector, TransportError, TransportResult, -}; -use aimdb_core::{AimDb, DbError, DbResult}; - -use crate::framing::{encode_frame, FrameAccumulator}; -use crate::DEFAULT_SCHEME; - -type BoxFuture = Pin + Send + 'static>>; -type BuildFuture<'a> = Pin>> + Send + 'a>>; - -/// How many bytes a single serial `read()` pulls before re-checking for a frame. -const READ_CHUNK: usize = 256; - -// =========================================================================== -// Connection -// =========================================================================== - -/// A framed bidirectional pipe over an async serial byte stream. Framing lives in -/// the transport: [`recv`](Connection::recv) returns one COBS frame (sentinel -/// stripped); [`send`](Connection::send) COBS-encodes and appends the sentinel. -pub struct TokioSerialConnection { - stream: S, - acc: FrameAccumulator, - peer: PeerInfo, -} - -impl TokioSerialConnection { - /// Wrap an already-open async byte stream (a `SerialStream`, or a duplex pipe - /// in tests). - pub fn new(stream: S) -> Self { - Self { - stream, - acc: FrameAccumulator::new(), - peer: PeerInfo::default(), - } - } -} - -impl Connection for TokioSerialConnection -where - S: AsyncRead + AsyncWrite + Unpin + Send, -{ - fn recv(&mut self) -> BoxFut<'_, TransportResult>>> { - Box::pin(async move { - loop { - // COBS is self-synchronizing: a chunk that fails to decode is line - // noise or a mid-stream join, not a fatal transport error. The - // accumulator has already consumed it, so skip it and resync on the - // next sentinel rather than tearing down the session. - match self.acc.next_frame() { - Some(Ok(frame)) => return Ok(Some(frame)), - Some(Err(_)) => continue, - None => {} - } - let mut chunk = [0u8; READ_CHUNK]; - match self.stream.read(&mut chunk).await { - Ok(0) => return Ok(None), // EOF — peer closed - Ok(n) => self.acc.push_bytes(&chunk[..n]), - Err(_) => return Err(TransportError::Io), - } - } - }) - } - - fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> { - Box::pin(async move { - let mut out = Vec::new(); - encode_frame(frame, &mut out); - self.stream - .write_all(&out) - .await - .map_err(|_| TransportError::Closed)?; - self.stream - .flush() - .await - .map_err(|_| TransportError::Closed) - }) - } - - fn peer(&self) -> &PeerInfo { - &self.peer - } -} - -// =========================================================================== -// Dialer / Listener -// =========================================================================== - -/// The initiating (client) side: opens the serial port on each -/// [`connect`](Dialer::connect). Cheap to clone (path + baud), so `run_client` -/// can redial and the generic `SessionClientConnector` can hold it. -#[derive(Clone)] -pub struct SerialDialer { - path: String, - baud: u32, -} - -impl SerialDialer { - /// Dial the serial device at `path` (e.g. `/dev/ttyUSB0`) at `baud`. - pub fn new(path: impl Into, baud: u32) -> Self { - Self { - path: path.into(), - baud, - } - } -} - -impl Dialer for SerialDialer { - fn connect(&self) -> BoxFut<'_, TransportResult>> { - Box::pin(async move { - let stream = tokio_serial::new(&self.path, self.baud) - .open_native_async() - .map_err(|_| TransportError::Io)?; - // Discard any bytes left in the OS input buffer by a previous session - // (e.g. a half-read reply from a killed client). Otherwise the first - // frame is a stale leftover that fails to decode and desyncs the stream - // until the next COBS sentinel — a transient `Internal` on the first - // call or two. - use tokio_serial::SerialPort; - let _ = stream.clear(tokio_serial::ClearBuffer::Input); - Ok(Box::new(TokioSerialConnection::new(stream)) as Box) - }) - } -} - -/// The accepting (server) side. Serial is point-to-point, so this is a one-shot -/// listener: the first [`accept`](Listener::accept) hands out the (already-open) -/// port; later calls park forever (there is only ever one peer on a UART). -pub struct SerialListener { - stream: Option, -} - -impl SerialListener { - /// Wrap an already-open serial port. - pub fn new(stream: SerialStream) -> Self { - Self { - stream: Some(stream), - } - } -} - -impl Listener for SerialListener { - fn accept(&mut self) -> BoxFut<'_, TransportResult>> { - Box::pin(async move { - match self.stream.take() { - Some(s) => Ok(Box::new(TokioSerialConnection::new(s)) as Box), - // Point-to-point: no second peer ever arrives. - None => core::future::pending().await, - } - }) - } -} - -// =========================================================================== -// Client sugar -// =========================================================================== - -/// Constructs a [`SessionClientConnector`] that dials an AimX peer over a serial -/// port. `SerialClient::new(path, baud)` is sugar; chain `.scheme(...)` / -/// `.with_config(...)` on the returned connector. -pub struct SerialClient; - -impl SerialClient { - /// Mirror records to/from the AimX peer reachable at serial `path` (scheme - /// defaults to [`DEFAULT_SCHEME`]). - // Sugar constructor: intentionally returns the generic connector, not `Self`. - #[allow(clippy::new_ret_no_self)] - pub fn new( - path: impl Into, - baud: u32, - ) -> SessionClientConnector { - SessionClientConnector::new(SerialDialer::new(path, baud), AimxCodec).scheme(DEFAULT_SCHEME) - } -} - -// =========================================================================== -// Server sugar -// =========================================================================== - -/// Accepts an AimX connection over a serial port and serves the full AimX -/// toolset. Register it via `with_connector` to let a host (or another board) -/// query this db over a UART. -pub struct SerialServer { - path: String, - baud: u32, - config: AimxConfig, - scheme: String, -} - -impl SerialServer { - /// Serve AimX over the serial device at `path` (e.g. `/dev/ttyUSB0`) at - /// `baud`, with the default read-only policy / limits. - pub fn new(path: impl Into, baud: u32) -> Self { - Self { - path: path.into(), - baud, - config: AimxConfig::uds_default(), - scheme: DEFAULT_SCHEME.to_string(), - } - } - - /// Use a prepared [`AimxConfig`] for the security policy / limits (the - /// `socket_path` / `socket_permissions` fields are unused over serial). - pub fn with_config(mut self, config: AimxConfig) -> Self { - self.config = config; - self - } - - /// Set the security policy (read-only vs. per-record-writable). - pub fn security_policy(mut self, policy: SecurityPolicy) -> Self { - self.config = self.config.security_policy(policy); - self - } - - /// Maximum live subscriptions per connection. - pub fn max_subs_per_connection(mut self, max: usize) -> Self { - self.config = self.config.max_subs_per_connection(max); - self - } - - /// Override the scheme this connector registers. - pub fn scheme(mut self, scheme: impl Into) -> Self { - self.scheme = scheme.into(); - self - } -} - -impl ConnectorBuilder for SerialServer { - fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { - let path = self.path.clone(); - let baud = self.baud; - let config = self.config.clone(); - let scheme = self.scheme.clone(); - Box::pin(async move { - let session_config = SessionConfig { - limits: SessionLimits { - // A UART carries a single peer; cap connections at 1. - max_connections: 1, - max_subs_per_connection: config.max_subs_per_connection, - }, - reads_hello: false, - // AimX's subscribe ack stays implicit (events flow); no ack frame. - acks_subscribe: false, - }; - let dispatch_config = config; - // Reuse the generic spine: open the port (errors surface synchronously) - // + AimX dispatch over the AimX codec. - let connector = SessionServerConnector::new( - move || open_serial_listener(&path, baud), - AimxCodec, - move |db: &AimDb| -> Arc { - // Apply the security policy's writable marking so `record.list` - // reports the `writable` flag (the dispatch also enforces it). - crate::apply_writable(db, &dispatch_config); - Arc::new(AimxDispatch::new( - Arc::new(db.clone()), - dispatch_config.clone(), - )) - }, - session_config, - ) - .scheme(scheme); - connector.build(db).await - }) - } - - fn scheme(&self) -> &str { - &self.scheme - } -} - -// =========================================================================== -// Helpers -// =========================================================================== - -/// Open the serial port synchronously so an open error surfaces from `build`. -fn open_serial_listener(path: &str, baud: u32) -> DbResult { - log_info!( - "Initializing AimX serial server on {} @ {} baud", - path, - baud - ); - - let stream = tokio_serial::new(path, baud) - .open_native_async() - .map_err(|e| DbError::IoWithContext { - context: format!("Failed to open serial port {} @ {} baud", path, baud), - source: std::io::Error::other(e), - })?; - Ok(SerialListener::new(stream)) -} diff --git a/aimdb-serial-connector/tests/embassy_smoke.rs b/aimdb-serial-connector/tests/embassy_smoke.rs index 869516a8..c0265e71 100644 --- a/aimdb-serial-connector/tests/embassy_smoke.rs +++ b/aimdb-serial-connector/tests/embassy_smoke.rs @@ -1,6 +1,6 @@ //! Embassy client-exit smoke — the runtime-neutral `run_client` engine drives RPC -//! over the **real** Embassy serial transport ([`SerialDialer`] / -//! `EmbassySerialConnection`, COBS over `embedded-io-async`) on the +//! over the **real** Embassy serial path (COBS framing over the adapter's +//! `EmbassyUart`) on the //! [`EmbassyAdapter`] clock. The `thumbv7em` monomorphization an MCU uses, driven //! on the host by `futures::executor::block_on` (no `embassy-executor`, which does //! not build on the host). @@ -11,7 +11,7 @@ //! [`EchoCodec`], so no second node is needed), exercising COBS encode → wire → //! decode under the engine. -#![cfg(feature = "embassy-runtime")] +#![cfg(feature = "_test-embassy")] extern crate alloc; @@ -24,12 +24,13 @@ use std::sync::Arc; use embedded_io_async::{ErrorKind, ErrorType, Read, Write}; +use aimdb_core::session::OneShotDialer; use aimdb_core::session::{ run_client, ClientConfig, CodecError, EnvelopeCodec, Inbound, Outbound, Payload, }; -use aimdb_embassy_adapter::connectors::{EmbassyConnection, OneShotDialer}; +use aimdb_embassy_adapter::io::EmbassyUart; use aimdb_embassy_adapter::EmbassyAdapter; -use aimdb_serial_connector::embassy_transport::CobsFramer; +use aimdb_serial_connector::connector::framed; // No-op defmt logger + host time driver so the binary links: the engine holds // the adapter as `Arc`, whose vtable references @@ -125,7 +126,7 @@ fn embassy_clock_drives_client_engine_rpc_over_serial() { use futures::executor::block_on; use futures::future::{select, Either}; - // The exact `run_client, _, EmbassyAdapter>` monomorphization + // The exact `run_client, _, EmbassyAdapter>` monomorphization // an MCU build uses — over the real COBS serial connection. let clock = Arc::new(EmbassyAdapter::default()); let config = ClientConfig { @@ -135,10 +136,9 @@ fn embassy_clock_drives_client_engine_rpc_over_serial() { }; let uart = LoopbackUart::default(); - // The one-shot dialer over the real COBS framed connection — the exact spine - // an MCU build uses (`OneShotDialer>`). - let conn = EmbassyConnection::<_, _, _>::new(uart.clone(), uart, CobsFramer::new()); - let dialer = OneShotDialer::new(conn); + // The one-shot dialer over the real COBS framed connection — the exact + // spine an MCU build uses. + let dialer = OneShotDialer::new(framed(EmbassyUart::new(uart.clone(), uart))); let (handle, engine_fut) = run_client(dialer, EchoCodec, config, clock); block_on(async move { diff --git a/aimdb-serial-connector/tests/framed.rs b/aimdb-serial-connector/tests/framed.rs index ad7eb64d..08f76080 100644 --- a/aimdb-serial-connector/tests/framed.rs +++ b/aimdb-serial-connector/tests/framed.rs @@ -1,21 +1,19 @@ //! The COBS framer and core's `FramedConnection` over the Tokio adapter's byte //! stream — the same pairing the Embassy side gets from `EmbassyUart`. -#![cfg(feature = "tokio-runtime")] +#![cfg(feature = "std")] use aimdb_core::session::Connection; -use aimdb_serial_connector::framing::{CobsFramer, TokioFramed, WRITE_CHUNK}; +use aimdb_serial_connector::connector::{framed, SerialFramed}; +use aimdb_serial_connector::framing::WRITE_CHUNK; use aimdb_tokio_adapter::net::TokioByteStream; /// A duplex pipe standing in for a `SerialStream`, framed at both ends. fn pipe() -> ( - TokioFramed, - TokioFramed, + SerialFramed>, + SerialFramed>, ) { let (a, b) = tokio::io::duplex(8 * 1024); - ( - TokioFramed::new(TokioByteStream(a), CobsFramer::new()), - TokioFramed::new(TokioByteStream(b), CobsFramer::new()), - ) + (framed(TokioByteStream(a)), framed(TokioByteStream(b))) } #[tokio::test] @@ -101,3 +99,187 @@ async fn a_boxed_connection_crosses_a_spawn() { ); echo.await.expect("echo task"); } + +// --------------------------------------------------------------------------- +// The neutral client/server sugar over an adapter byte stream. +// --------------------------------------------------------------------------- + +/// A UART is point-to-point: the stream is served once, so a second `accept` +/// parks rather than erroring — `serve` would otherwise spin on it. +#[tokio::test] +async fn a_one_shot_listener_yields_once_then_parks() { + use aimdb_core::session::Listener; + use aimdb_core::session::OneShotListener; + use aimdb_serial_connector::connector::framed; + + let (a, _b) = tokio::io::duplex(1024); + let mut listener = OneShotListener::new(framed(TokioByteStream(a))); + + assert!(listener.accept().await.is_ok(), "first accept yields"); + + let parked = + tokio::time::timeout(std::time::Duration::from_millis(50), listener.accept()).await; + assert!(parked.is_err(), "a second accept must park, not resolve"); +} + +/// The dialer's dual: nothing to redial on a UART, so a second attempt is a +/// real error rather than a silent reconnect loop. +#[tokio::test] +async fn a_one_shot_dialer_refuses_a_second_connect() { + use aimdb_core::session::Dialer; + use aimdb_core::session::OneShotDialer; + use aimdb_serial_connector::connector::framed; + + let (a, _b) = tokio::io::duplex(1024); + let dialer = OneShotDialer::new(framed(TokioByteStream(a))); + + assert!(dialer.connect().await.is_ok(), "first connect yields"); + assert!( + dialer.connect().await.is_err(), + "a UART has no second connection to hand out" + ); +} + +/// The stream is moved in, so a second `build` is refused, and a `build()` +/// future dropped before it is polled must leave the stream in place. +#[tokio::test] +async fn the_server_guards_its_moved_in_stream() { + use aimdb_core::buffer::BufferCfg; + use aimdb_core::connector::ConnectorBuilder; + use aimdb_core::AimDbBuilder; + use aimdb_serial_connector::connector::SerialServer; + use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + + let mut builder = AimDbBuilder::new().runtime(std::sync::Arc::new(TokioAdapter)); + builder.configure::("counter", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (db, _runner) = builder.build().await.expect("build db"); + + let (a, _b) = tokio::io::duplex(1024); + let server = SerialServer::new(TokioByteStream(a)); + + // Unpolled build: the stream must survive it. + drop(server.build(&db)); + + let futures = server + .build(&db) + .await + .expect("the stream must survive an unpolled build"); + assert_eq!(futures.len(), 1, "one serve future"); + + let Err(err) = server.build(&db).await else { + panic!("a second build must fail"); + }; + assert!( + format!("{err}").contains("already taken"), + "unexpected error: {err}" + ); +} + +/// The client sugar must actually be registrable: `SessionClientConnector` only +/// implements `ConnectorBuilder` when its dialer satisfies the bounds, and a +/// dialer wrapping a moved-in stream cannot be `Clone`. Nothing else in the repo +/// calls `SerialClient::new`, so without this the constructor can stop +/// compiling at its use site while every leg stays green. +#[tokio::test] +async fn a_serial_client_is_a_registrable_connector() { + use aimdb_core::connector::ConnectorBuilder; + use aimdb_serial_connector::connector::SerialClient; + + fn assert_builder(_: T) {} + + let (a, _b) = tokio::io::duplex(1024); + assert_builder(SerialClient::new(TokioByteStream(a))); +} + +/// A moved-in stream cannot be re-acquired, so the client engine must end when +/// the peer goes away rather than redial a dialer that can never succeed again. +/// +/// Two independent things guarantee that, and this pins their conjunction: +/// `SerialClient::new` sets `reconnect: false`, and `run_client` treats a +/// dialer's `Closed` as terminal. Drop either and the test still passes; drop +/// both — `reconnect: true` with the dialer reporting `Io` — and it hangs until +/// the timeout, which is what a board would do forever. +#[tokio::test] +async fn a_one_shot_client_stops_instead_of_redialing_forever() { + use aimdb_core::buffer::BufferCfg; + use aimdb_core::connector::ConnectorBuilder; + use aimdb_core::AimDbBuilder; + use aimdb_serial_connector::connector::{SerialClient, SerialServer}; + use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + + // An outbound route is what keeps a `ClientHandle` alive past `build`; + // without one every sender drops there and the engine ends on its own, + // whatever the reconnect policy says. The route needs a connector + // registered under its scheme to pass validation, so a `SerialServer` on a + // duplex nobody talks to stands in — the client under test is built by hand + // below so its futures stay reachable. + let (server_end, _server_peer) = tokio::io::duplex(1024); + let mut builder = AimDbBuilder::new() + .runtime(std::sync::Arc::new(TokioAdapter)) + .with_connector(SerialServer::new(TokioByteStream(server_end))); + builder.configure::("counter", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .with_remote_access() + .link_to("serial://counter") + .with_serializer(|_ctx, v: &u64| Ok(v.to_le_bytes().to_vec())) + .finish(); + }); + let (db, _runner) = builder.build().await.expect("build db"); + + let (a, b) = tokio::io::duplex(1024); + let client = SerialClient::new(TokioByteStream(a)); + + let mut futures = client.build(&db).await.expect("build the connector"); + assert_eq!( + futures.len(), + 2, + "one outbound pump plus the engine — the pump is what holds the handle" + ); + // `SessionClientConnector::build` pushes the engine after the pumps. + let engine = futures.pop().expect("engine future"); + + // Peer hangs up: the one connection this dialer had is gone for good. + drop(b); + + tokio::time::timeout(std::time::Duration::from_secs(2), engine) + .await + .expect("the engine must end, not redial a stream that cannot be reopened"); +} + +/// The dialer is moved in, so a second `build` is refused rather than handing +/// out a connection that was already consumed. +#[tokio::test] +async fn the_client_guards_its_moved_in_dialer() { + use aimdb_core::buffer::BufferCfg; + use aimdb_core::connector::ConnectorBuilder; + use aimdb_core::AimDbBuilder; + use aimdb_serial_connector::connector::SerialClient; + use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; + + let mut builder = AimDbBuilder::new().runtime(std::sync::Arc::new(TokioAdapter)); + builder.configure::("counter", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (db, _runner) = builder.build().await.expect("build db"); + + let (a, _b) = tokio::io::duplex(1024); + let client = SerialClient::new(TokioByteStream(a)); + + // Unpolled build: the dialer must survive it. + drop(client.build(&db)); + + client + .build(&db) + .await + .expect("the dialer must survive an unpolled build"); + + let Err(err) = client.build(&db).await else { + panic!("a second build must fail"); + }; + assert!( + format!("{err}").contains("already taken"), + "unexpected error: {err}" + ); +} diff --git a/aimdb-serial-connector/tests/tokio_roundtrip.rs b/aimdb-serial-connector/tests/tokio_roundtrip.rs index fbe5f03d..dcdb3ebc 100644 --- a/aimdb-serial-connector/tests/tokio_roundtrip.rs +++ b/aimdb-serial-connector/tests/tokio_roundtrip.rs @@ -1,28 +1,27 @@ //! End-to-end AimX over the tokio serial transport, without hardware: the two //! ends of a `tokio::io::duplex()` pipe stand in for a crossover serial cable. //! The production server (`serve` + `AimxDispatch`) answers on one end; the -//! `run_client` engine drives RPC on the other — proving `TokioSerialConnection`'s -//! COBS framing carries the real protocol both directions. +//! `run_client` engine drives RPC on the other — proving the COBS framing over +//! the adapter's byte stream carries the real protocol both directions. -#![cfg(feature = "_test-tokio")] +#![cfg(feature = "std")] use std::sync::Arc; -use std::sync::Mutex; use std::time::Duration; use aimdb_core::buffer::BufferCfg; use aimdb_core::remote::AimxConfig; use aimdb_core::session::aimx::{AimxCodec, AimxDispatch}; use aimdb_core::session::{ - run_client, serve, BoxFut, ClientConfig, Connection, Dialer, Dispatch, Listener, Payload, - SessionConfig, SessionLimits, TransportError, TransportResult, + run_client, serve, ClientConfig, Connection, Dispatch, Payload, SessionConfig, SessionLimits, }; +use aimdb_core::session::{OneShotDialer, OneShotListener}; use aimdb_core::AimDbBuilder; -use aimdb_serial_connector::tokio_transport::TokioSerialConnection; +use aimdb_serial_connector::connector::framed; +use aimdb_tokio_adapter::net::TokioByteStream; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use serde::{Deserialize, Serialize}; use serde_json::json; -use tokio::io::DuplexStream; /// A writable config-style record (SingleLatest, no producer → remotely settable). #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -30,32 +29,6 @@ struct Setting { level: u64, } -/// One-shot dialer over an in-memory duplex end (stands in for opening the port). -struct OnceDialer(Mutex>); - -impl Dialer for OnceDialer { - fn connect(&self) -> BoxFut<'_, TransportResult>> { - Box::pin(async move { - let end = self.0.lock().unwrap().take().ok_or(TransportError::Io)?; - Ok(Box::new(TokioSerialConnection::new(end)) as Box) - }) - } -} - -/// One-shot listener over the other duplex end (point-to-point, like a UART). -struct OnceListener(Option); - -impl Listener for OnceListener { - fn accept(&mut self) -> BoxFut<'_, TransportResult>> { - Box::pin(async move { - match self.0.take() { - Some(end) => Ok(Box::new(TokioSerialConnection::new(end)) as Box), - None => core::future::pending().await, - } - }) - } -} - #[tokio::test] async fn aimx_roundtrips_over_the_serial_transport() { let (server_end, client_end) = tokio::io::duplex(8192); @@ -84,7 +57,7 @@ async fn aimx_roundtrips_over_the_serial_transport() { acks_subscribe: false, }; tokio::spawn(serve( - OnceListener(Some(server_end)), + OneShotListener::new(framed(TokioByteStream(server_end))), Arc::new(AimxCodec), dispatch, session_config, @@ -96,7 +69,7 @@ async fn aimx_roundtrips_over_the_serial_transport() { ..ClientConfig::default() }; let (handle, engine) = run_client( - OnceDialer(Mutex::new(Some(client_end))), + OneShotDialer::new(framed(TokioByteStream(client_end))), AimxCodec, client_config, Arc::new(TokioAdapter), @@ -139,7 +112,7 @@ async fn recv_resyncs_past_a_corrupt_frame() { use tokio::io::AsyncWriteExt; let (mut peer, conn_end) = tokio::io::duplex(1024); - let mut conn = TokioSerialConnection::new(conn_end); + let mut conn = framed(TokioByteStream(conn_end)); // `0x05` is a COBS code byte promising four more bytes that never arrive, so the // delimited chunk fails to decode; follow it with a valid frame. diff --git a/docs/design/052-runtime-neutral-connectors.md b/docs/design/052-runtime-neutral-connectors.md index c644242d..d8db4cef 100644 --- a/docs/design/052-runtime-neutral-connectors.md +++ b/docs/design/052-runtime-neutral-connectors.md @@ -193,7 +193,7 @@ Three details the prototype settled: - **`aimdb-embassy-adapter`**: `EmbassyNet::tcp(stack, rx, tx)`, `EmbassyNet::listen::(stack, endpoint, rx[N], tx[N])` (the socket-slot pool moves here from `aimdb-tcp-connector/src/embassy_transport.rs`, since deleted), - `EmbassyNet::udp(stack, …)`, `EmbassyUart::split(rx, tx)`, and `Delay` + `EmbassyNet::udp(stack, …)`, `EmbassyUart::new(rx, tx)`, and `Delay` returning `embassy_time::Timer`. Each stream/datagram newtype is `unsafe impl Send` and wraps the inner future in `SendFutureWrapper`. The `unsafe` stays exactly where design 033 put it. `NetStack` construction moves @@ -579,7 +579,8 @@ let mqtt = MqttConnector::new(broker_url).transport(TokioNet::tcp()); ### 7.2 Embassy ```rust -use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyNet, EmbassyUart}; +use aimdb_embassy_adapter::io::EmbassyUart; +use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyNet}; use aimdb_mqtt_connector::MqttConnector; #[cfg(feature = "tls")] use aimdb_mqtt_connector::embedded::TlsOptions; @@ -610,7 +611,7 @@ let mqtt = mqtt.with_tls(TlsOptions::new( // CHANGED: the UART halves go through the adapter too, instead of the // connector's `embassy_transport` module. let (serial_tx, serial_rx) = uart.split(); -let serial = SerialServer::new(EmbassyUart::split(serial_rx, serial_tx)) +let serial = SerialServer::new(EmbassyUart::new(serial_rx, serial_tx)) .security_policy(SecurityPolicy::read_only()); let mut builder = AimDbBuilder::new() @@ -637,8 +638,10 @@ What moved: RNG that does not will see the error at this line rather than a stray `unsafe impl` deep in the connector, which is the point. - **The serial UART goes through the adapter.** `SerialServer::new(rx, tx)` - becomes `SerialServer::new(EmbassyUart::split(rx, tx))`, so the connector - names no `embedded-io-async` halves of its own. + becomes `SerialServer::new(EmbassyUart::new(rx, tx))`, so the connector names + no `embedded-io-async` halves of its own. `EmbassyUart` lives in the adapter's + `io` module, behind `connector-io` — it borrows nothing from `embassy-net`, so + a serial-only board compiles no network stack to frame a UART. Behind the scenes the type is `MqttConnector>` for the `.transport(...)` path and `MqttConnector` for the rumqttc path. diff --git a/examples/embassy-knx-connector-demo/Cargo.toml b/examples/embassy-knx-connector-demo/Cargo.toml index 1e0c2956..7a75b043 100644 --- a/examples/embassy-knx-connector-demo/Cargo.toml +++ b/examples/embassy-knx-connector-demo/Cargo.toml @@ -17,6 +17,8 @@ aimdb-core = { path = "../../aimdb-core", default-features = false, features = [ ] } aimdb-embassy-adapter = { path = "../../aimdb-embassy-adapter", default-features = false, features = [ "embassy-runtime", + # `EmbassyUart`, the byte source the serial connector frames. + "connector-io", ] } aimdb-knx-connector = { path = "../../aimdb-knx-connector", default-features = false, features = [ "embassy-runtime", @@ -24,9 +26,9 @@ aimdb-knx-connector = { path = "../../aimdb-knx-connector", default-features = f ] } # Serial remote-access server — serves this db's records over a UART (ST-LINK VCP) # so a host can read them with `aimdb --features transport-serial --connect serial://…`. -# Its `embassy-runtime` feature also turns on `aimdb-core/remote`. +# Its `connector` feature also turns on `aimdb-core/remote`. aimdb-serial-connector = { path = "../../aimdb-serial-connector", default-features = false, features = [ - "embassy-runtime", + "connector", "defmt", ] } diff --git a/examples/embassy-knx-connector-demo/src/main.rs b/examples/embassy-knx-connector-demo/src/main.rs index 84e4c032..aee7ad7e 100644 --- a/examples/embassy-knx-connector-demo/src/main.rs +++ b/examples/embassy-knx-connector-demo/src/main.rs @@ -42,10 +42,11 @@ extern crate alloc; use aimdb_core::remote::SecurityPolicy; use aimdb_core::{AimDbBuilder, RecordKey, RuntimeContext}; +use aimdb_embassy_adapter::io::EmbassyUart; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; use aimdb_knx_connector::dpt::{Dpt1, Dpt9, DptDecode, DptEncode}; use aimdb_knx_connector::embassy_client::KnxConnectorBuilder; -use aimdb_serial_connector::embassy_transport::SerialServer; +use aimdb_serial_connector::SerialServer; use defmt::*; use embassy_executor::Spawner; use embassy_net::StackResources; @@ -275,7 +276,8 @@ async fn main(spawner: Spawner) { .runtime(runtime.clone()) .with_connector(KnxConnectorBuilder::new(&gateway_url, stack)) .with_connector( - SerialServer::new(serial_rx, serial_tx).security_policy(SecurityPolicy::read_only()), + SerialServer::new(EmbassyUart::new(serial_rx, serial_tx)) + .security_policy(SecurityPolicy::read_only()), ); // ======================================================================== diff --git a/examples/embassy-mqtt-connector-demo/Cargo.toml b/examples/embassy-mqtt-connector-demo/Cargo.toml index 6775c259..0a4725f3 100644 --- a/examples/embassy-mqtt-connector-demo/Cargo.toml +++ b/examples/embassy-mqtt-connector-demo/Cargo.toml @@ -21,15 +21,17 @@ std = [] aimdb-core = { path = "../../aimdb-core", default-features = false } aimdb-embassy-adapter = { path = "../../aimdb-embassy-adapter", default-features = false, features = [ "embassy-runtime", + # `EmbassyUart`, the byte source the serial connector frames. + "connector-io", ] } aimdb-mqtt-connector = { path = "../../aimdb-mqtt-connector", default-features = false, features = [ "embassy-runtime", ] } # Serial remote-access server — serves this db's records over a UART (ST-LINK VCP) # so a host can read them with `aimdb --features transport-serial --connect serial://…`. -# Its `embassy-runtime` feature also turns on `aimdb-core/remote`. +# Its `connector` feature also turns on `aimdb-core/remote`. aimdb-serial-connector = { path = "../../aimdb-serial-connector", default-features = false, features = [ - "embassy-runtime", + "connector", "defmt", ] } diff --git a/examples/embassy-mqtt-connector-demo/src/main.rs b/examples/embassy-mqtt-connector-demo/src/main.rs index abd996c8..463feb6b 100644 --- a/examples/embassy-mqtt-connector-demo/src/main.rs +++ b/examples/embassy-mqtt-connector-demo/src/main.rs @@ -75,8 +75,9 @@ extern crate alloc; use aimdb_core::remote::SecurityPolicy; use aimdb_core::{AimDbBuilder, Producer, RecordKey, RuntimeContext}; +use aimdb_embassy_adapter::io::EmbassyUart; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; -use aimdb_serial_connector::embassy_transport::SerialServer; +use aimdb_serial_connector::SerialServer; use defmt::*; use embassy_executor::Spawner; use embassy_net::StackResources; @@ -410,7 +411,8 @@ async fn main(spawner: Spawner) { .runtime(runtime.clone()) .with_connector(mqtt) .with_connector( - SerialServer::new(serial_rx, serial_tx).security_policy(SecurityPolicy::read_only()), + SerialServer::new(EmbassyUart::new(serial_rx, serial_tx)) + .security_policy(SecurityPolicy::read_only()), ); // ======================================================================== diff --git a/examples/embassy-serial-connector-demo/Cargo.toml b/examples/embassy-serial-connector-demo/Cargo.toml index 2354e4bd..27062767 100644 --- a/examples/embassy-serial-connector-demo/Cargo.toml +++ b/examples/embassy-serial-connector-demo/Cargo.toml @@ -17,9 +17,11 @@ aimdb-core = { path = "../../aimdb-core", default-features = false, features = [ ] } aimdb-embassy-adapter = { path = "../../aimdb-embassy-adapter", default-features = false, features = [ "embassy-runtime", + # `EmbassyUart`, the byte source the serial connector frames. + "connector-io", ] } aimdb-serial-connector = { path = "../../aimdb-serial-connector", default-features = false, features = [ - "embassy-runtime", + "connector", "defmt", ] } diff --git a/examples/embassy-serial-connector-demo/src/main.rs b/examples/embassy-serial-connector-demo/src/main.rs index 7ad33b52..c5ee2288 100644 --- a/examples/embassy-serial-connector-demo/src/main.rs +++ b/examples/embassy-serial-connector-demo/src/main.rs @@ -33,8 +33,9 @@ extern crate alloc; use aimdb_core::remote::SecurityPolicy; use aimdb_core::{AimDbBuilder, Producer}; +use aimdb_embassy_adapter::io::EmbassyUart; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; -use aimdb_serial_connector::embassy_transport::SerialServer; +use aimdb_serial_connector::SerialServer; use defmt::*; use embassy_executor::Spawner; use embassy_stm32::usart::{BufferedUart, Config as UartConfig}; @@ -144,7 +145,7 @@ async fn main(spawner: Spawner) { let runtime = alloc::sync::Arc::new(EmbassyAdapter::default()); let mut builder = AimDbBuilder::new() .runtime(runtime) - .with_connector(SerialServer::new(rx, tx).security_policy(policy)); + .with_connector(SerialServer::new(EmbassyUart::new(rx, tx)).security_policy(policy)); builder.configure::("counter", |reg| { reg.buffer_sized::<4, 2>(EmbassyBufferType::SingleLatest) .with_remote_access();