From e8bbc7465f463a781e2e9ea545c43743dc266be1 Mon Sep 17 00:00:00 2001 From: "glm-5.2" Date: Sun, 12 Jul 2026 11:07:17 +0000 Subject: [PATCH] refactor(core): add Connection::from_source public constructor (ADR-070 gap) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The BidiStreamSource trait (ADR-070) made Connection hold Box so downstream crates can add connection shapes without editing core — but the public constructor for that path was missing. The source field is private; the channels crate (or any future crate) had no way to build a Connection from its own BidiStreamSource impl. The trait was unusable from outside core. Add Connection::from_source(source: impl BidiStreamSource, alpn: Vec) -> Self: the extension point. Takes impl BidiStreamSource (not Box) to match the ergonomic style of from_stream / from_bidi; boxes the source internally. No feature gate (always available, like from_stream). Initializes alpn and identity: OnceLock::new() the same as the other constructors. Test: a custom RecordingSource BidiStreamSource impl (not a built-in) constructed via from_source, verifying remote_alpn / remote_addr / accept_bi (round-trips real bytes via tokio::io::duplex) / open_bi (StreamClosed) / close (records code+reason) all delegate to the custom impl. Verified: cargo test + clippy clean across default, --no-default-features, --features iroh, --all-features; cargo check -p alknet-call succeeds unchanged; cargo fmt --check passes. --- crates/alknet-core/src/types.rs | 123 ++++++++++++++++++ .../connection-from-source-constructor.md | 2 +- 2 files changed, 124 insertions(+), 1 deletion(-) diff --git a/crates/alknet-core/src/types.rs b/crates/alknet-core/src/types.rs index 207b677..6dadc00 100644 --- a/crates/alknet-core/src/types.rs +++ b/crates/alknet-core/src/types.rs @@ -568,6 +568,17 @@ impl Connection { Self::from_stream(send, recv, alpn, remote_addr) } + /// Construct from a caller-supplied `BidiStreamSource` impl. The + /// extension point for downstream crates — implement the trait and + /// construct a `Connection` from it without editing core. See ADR-070. + pub fn from_source(source: impl BidiStreamSource, alpn: Vec) -> Self { + Self { + source: Box::new(source), + alpn, + identity: OnceLock::new(), + } + } + /// Yield the next bidirectional stream this connection provides. /// /// # Transport semantics @@ -636,6 +647,118 @@ fn map_iroh_connection_error(e: iroh::endpoint::ConnectionError) -> StreamError } } +#[cfg(test)] +mod from_source_tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + use std::sync::Arc; + + /// A minimal custom `BidiStreamSource` impl to prove `from_source` + /// delegates to a caller-supplied impl. Not a built-in — the whole + /// point of `from_source` is that a non-core type can drive `Connection`. + struct RecordingSource { + stream: Mutex>, + addr: Option, + closed: Arc>>, + } + + #[async_trait] + impl BidiStreamSource for RecordingSource { + async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> { + match self.stream.lock().expect("mock mutex poisoned").take() { + Some(pair) => Ok(pair), + None => Err(StreamError::ConnectionClosed), + } + } + + async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> { + Err(StreamError::StreamClosed) + } + + fn remote_addr(&self) -> Option { + self.addr + } + + fn close(&self, code: u32, reason: &str) { + let _ = self + .closed + .lock() + .expect("mock closed mutex poisoned") + .replace((code, reason.to_string())); + } + } + + #[tokio::test] + async fn from_source_delegates_to_custom_impl() { + use tokio::io::AsyncReadExt; + use tokio::io::AsyncWriteExt; + + // One duplex: the mock holds end `a` (split into send_a/recv_a); the + // test driver holds end `b` (split into send_b/recv_b) to echo back. + let (a, b) = tokio::io::duplex(64); + let (recv_a, send_a) = tokio::io::split(a); + let (mut recv_b, mut send_b) = tokio::io::split(b); + let addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 7777)); + let recorded = Arc::new(Mutex::new(None)); + let conn = Connection::from_source( + RecordingSource { + stream: Mutex::new(Some(( + SendStream::from_stream(send_a), + RecvStream::from_stream(recv_a), + ))), + addr, + closed: Arc::clone(&recorded), + }, + b"alknet/test".to_vec(), + ); + + // remote_alpn reads self.alpn (Connection-level, unchanged). + assert_eq!(conn.remote_alpn(), b"alknet/test"); + + // remote_addr delegates to RecordingSource::remote_addr. + assert_eq!(conn.remote_addr(), addr); + + // accept_bi delegates to RecordingSource::accept_bi and yields the pair. + let (mut send, mut recv) = conn.accept_bi().await.expect("first accept_bi yields"); + + // Write via the mock's SendStream -> arrives at the driver's recv_b. + send.write_all(b"hello").await.expect("write round-trips"); + let mut buf = [0u8; 5]; + recv_b.read_exact(&mut buf).await.expect("driver reads"); + assert_eq!(&buf, b"hello"); + + // Driver writes back -> arrives at the mock's RecvStream. + send_b + .write_all(b"world") + .await + .expect("driver writes back"); + let mut buf = [0u8; 5]; + recv.read_exact(&mut buf).await.expect("read round-trips"); + assert_eq!(&buf, b"world"); + + // Second accept_bi delegates to RecordingSource::accept_bi -> ConnectionClosed. + match conn.accept_bi().await { + Err(StreamError::ConnectionClosed) => {} + Err(e) => panic!("expected ConnectionClosed on second accept_bi, got {e}"), + Ok(_) => panic!("expected ConnectionClosed on second accept_bi, got a stream"), + } + + // open_bi delegates to RecordingSource::open_bi -> StreamClosed. + match conn.open_bi().await { + Err(StreamError::StreamClosed) => {} + Err(e) => panic!("expected StreamClosed from open_bi, got {e}"), + Ok(_) => panic!("expected StreamClosed from open_bi, got a stream"), + } + + // close delegates to RecordingSource::close and the args thread through. + conn.close(42, "shutting down"); + assert_eq!( + recorded.lock().expect("recorded mutex poisoned").take(), + Some((42, "shutting down".to_string())) + ); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/tasks/core/connection-from-source-constructor.md b/tasks/core/connection-from-source-constructor.md index c4ff652..40703af 100644 --- a/tasks/core/connection-from-source-constructor.md +++ b/tasks/core/connection-from-source-constructor.md @@ -1,7 +1,7 @@ --- id: core/connection-from-source-constructor name: "Add Connection::from_source constructor (ADR-070 gap — the public extension point)" -status: pending +status: completed depends_on: [] scope: narrow risk: low