alkdev/alknet
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(core,http,tty,call): unify stream leaf — BiStream as the handler leaf (Phase 6)

Browse Source 
Implement ADR-092 across the workspace: accept_bi/open_bi return BiStream
(a concrete AsyncRead + AsyncWrite + Send + Unpin newtype), not the split
(SendStream, RecvStream) pair. The join moves into core's BidiStreamSource
impls (quinn/iroh via tokio::io::join, single-stream via boxed AsyncReadWrite);
handlers receive the joined BiStream and never see the pair.

Core (alknet-core/src/types.rs):
- Add concrete BiStream struct boxing Box<dyn AsyncReadWrite + Unpin>,
  with AsyncRead + AsyncWrite impls. from_joined (pub, for downstream
  crates that produce split halves naturally — channels reassembly, tests)
  and from_bidi (pub(crate), for Connection::from_bidi) constructors.
- Change BidiStreamSource::accept_bi/open_bi return types from
  (SendStream, RecvStream) to BiStream. Update QuinnBidiStreamSource,
  IrohBidiStreamSource, StreamBidiStreamSource impls to do the join once.
- Collapse SendStream/RecvStream to thin newtypes over
  Box<dyn Async* + Send + Unpin>. Remove SendStreamKind/RecvStreamKind
  enums and the quinn/iroh per-call dispatch (the join happens once in the
  BidiStreamSource impl now). Keep SendStream::from_stream /
  RecvStream::from_stream per-half boxing for into_sub_streams() (ADR-074)
  and the future channels reassembly path.
- Remove Connection::from_stream (split-pair constructor). Promote
  Connection::from_bidi to the only public stream constructor (the rule:
  the split never crosses a crate boundary as part of a constructor).
- Update Connection::accept_bi/open_bi to return BiStream. Update
  from_source_tests and tests modules to use from_bidi and BiStream;
  add a SinkEmpty test helper (AsyncRead EOF + AsyncWrite discard) for
  Connection-level-only test connections.

alknet-http (server/adapter.rs):
- Drop the 44-line QuicStream wrapper — accept_bi returns BiStream which
  is already AsyncRead + AsyncWrite. HttpAdapter::handle becomes 4 lines.
- Drop the 38-line QuicStreamDuplex test helper — tests use a single
  tokio::io::duplex whose ends are each AsyncRead + AsyncWrite natively.
- Remove unused std::io / std::pin::Pin imports.

alknet-tty (adapter.rs):
- TtyAdapter::handle splits the BiStream from accept_bi via
  tokio::io::split for drive_session's separate AsyncWrite/AsyncRead args
  (the stdlib idiom for TcpStream-style duplex streams).

alknet-call (protocol/*, client/*):
- Dispatcher::run_loop accept_bi site: take BiStream, pass to handle_stream.
- Dispatcher::handle_stream signature: take BiStream, split internally via
  tokio::io::split (was: take SendStream + RecvStream separately).
- CallConnection::call_with_payload / subscribe_with_payload / write_envelope:
  split the BiStream from open_bi via tokio::io::split at the call site.
- write_request / read_stream_until_closed: generic over AsyncWrite/AsyncRead
  (were: concrete SendStream/RecvStream) — accepts the ReadHalf/WriteHalf
  from tokio::io::split directly.
- Add protocol/test_support.rs with sink_empty_connection() (replaces the
  5 duplicated stub_connection() fns that used Connection::from_stream).
- Update all test stubs (call_client.rs, protocol/connection.rs,
  protocol/dispatch.rs, protocol/adapter.rs, client/from_call.rs) to use
  Connection::from_bidi + the shared sink_empty_connection() helper.
- Test handle_stream call sites: build BiStream::from_joined(recv, send)
  from the existing BufReader<Cursor> + duplex pair.

Workspace test status: all 9 crates pass (116 + 307 + 18 + 3 + 17 + 301 +
34 + 61 + 23 + 5 + 6 + 8 + 82 + 4 + 3 + 6 + 12 + 1 = 1007 tests pass). One
pre-existing failure remains in alknet-http
(adapters::to_mcp::tests::search_returns_access_control_filtered_ops_excluding_subscriptions
— handler kind mismatch, unrelated to Phase 6, fails on develop baseline).
This commit is contained in:
glm-5.2 committed 2026-07-18 15:59:01 +00:00
1 parent 249370345f
commit b60a5844ba
10 files changed
+412 -403
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/alknet-call/src/client/call_client.rs
+1 -9
View File
@@ -85,16 +85,8 @@ mod tests {
use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
use alknet_core::auth::Identity;
use alknet_core::types::Capabilities;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
fn stub_connection() -> Connection {
Connection::from_stream(
tokio::io::sink(),
tokio::io::empty(),
b"alknet/call".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321)),
)
}
use crate::protocol::sink_empty_connection as stub_connection;
fn external_spec(name: &str) -> OperationSpec {
OperationSpec::new(
crates/alknet-call/src/client/from_call.rs
+8 -28
View File
@@ -128,12 +128,9 @@ fn build_bundles(
Arc::new(op_summary.connection.clone()),
remote_name,
)),
OperationType::Query | OperationType::Mutation => {
HandlerKind::Once(make_forwarding_handler(
Arc::new(op_summary.connection.clone()),
remote_name,
))
}
OperationType::Query | OperationType::Mutation => HandlerKind::Once(
make_forwarding_handler(Arc::new(op_summary.connection.clone()), remote_name),
),
};
bundles.push(HandlerRegistration::new(
spec,
@@ -331,17 +328,13 @@ fn parse_access_control(v: &Value) -> AccessControl {
/// If `context.identity` is `None` (the hub chose not to disclose, or has not
/// authenticated an originator), `forwarded_for` is omitted — the spoke
/// receives only the hub's identity.
fn make_forwarding_handler(
connection: Arc<CallConnection>,
remote_name: String,
) -> Handler {
fn make_forwarding_handler(connection: Arc<CallConnection>, remote_name: String) -> Handler {
use crate::registry::registration::make_handler;
make_handler(move |input, context| {
let connection = Arc::clone(&connection);
let remote_name = remote_name.clone();
async move {
let payload =
build_forwarded_payload(&remote_name, input, &context);
let payload = build_forwarded_payload(&remote_name, input, &context);
// The forwarding handler invokes the remote op via the
// CallConnection. The parent_request_id participates in the abort
// cascade (ADR-016 §6): if the parent is aborted, the cascade
@@ -381,8 +374,7 @@ fn make_streaming_forwarding_handler(
let connection = Arc::clone(&connection);
let remote_name = remote_name.clone();
once(async move {
let payload =
build_forwarded_payload(&remote_name, input, &context);
let payload = build_forwarded_payload(&remote_name, input, &context);
connection.subscribe_with_payload(payload).await
})
.flatten()
@@ -393,11 +385,7 @@ fn make_streaming_forwarding_handler(
/// `forwarded_for` from the hub's `OperationContext.identity` (ADR-032 §3).
/// `forwarded_for` is omitted when `context.identity` is `None` (the hub
/// chooses not to disclose the originator).
fn build_forwarded_payload(
operation_id: &str,
input: Value,
context: &OperationContext,
) -> Value {
fn build_forwarded_payload(operation_id: &str, input: Value, context: &OperationContext) -> Value {
let mut payload = serde_json::Map::new();
payload.insert(
"operationId".to_string(),
@@ -421,17 +409,9 @@ mod tests {
use alknet_core::auth::Identity;
use alknet_core::types::Capabilities;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Mutex as StdMutex;
fn stub_connection() -> alknet_core::types::Connection {
alknet_core::types::Connection::from_stream(
tokio::io::sink(),
tokio::io::empty(),
b"alknet/call".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321)),
)
}
use crate::protocol::sink_empty_connection as stub_connection;
fn sample_schema_json(name: &str, op_type: &str) -> Value {
json!({
crates/alknet-call/src/protocol/adapter.rs
+8 -19
View File
@@ -140,10 +140,9 @@ impl CallAdapter {
pub(crate) async fn handle_stream(
&self,
connection: Arc<CallConnection>,
send: alknet_core::types::SendStream,
recv: alknet_core::types::RecvStream,
stream: alknet_core::types::BiStream,
) {
self.dispatcher.handle_stream(connection, send, recv).await;
self.dispatcher.handle_stream(connection, stream).await;
}
}
@@ -179,10 +178,11 @@ mod tests {
use alknet_core::auth::AuthToken;
use alknet_core::types::Capabilities;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Mutex as StdMutex;
use std::time::{Duration, Instant};
use crate::protocol::sink_empty_connection as stub_connection;
struct StaticIdentityProvider {
tokens: StdMutex<HashMap<String, Identity>>,
}
@@ -290,15 +290,6 @@ mod tests {
})
}
fn stub_connection() -> Connection {
Connection::from_stream(
tokio::io::sink(),
tokio::io::empty(),
b"alknet/call".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321)),
)
}
#[test]
fn alpn_returns_alknet_call() {
let registry = Arc::new(OperationRegistry::new());
@@ -1193,10 +1184,9 @@ mod tests {
let frame = encode_frame(&EventEnvelope::aborted("parent-1"));
let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame));
let (send, _recv_sink) = tokio::io::duplex(64);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
adapter.handle_stream(conn.clone(), send, recv).await;
adapter.handle_stream(conn.clone(), stream).await;
let pending = conn.pending().lock();
assert!(
@@ -1233,10 +1223,9 @@ mod tests {
let frame = encode_frame(&EventEnvelope::aborted("does-not-exist"));
let recv = tokio::io::BufReader::new(std::io::Cursor::new(frame));
let (send, _recv_sink) = tokio::io::duplex(64);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
adapter.handle_stream(conn.clone(), send, recv).await;
adapter.handle_stream(conn.clone(), stream).await;
let pending = conn.pending().lock();
assert!(
crates/alknet-call/src/protocol/connection.rs
+27 -21
View File
@@ -16,6 +16,7 @@ use alknet_core::types::Connection;
use futures::stream::Stream;
use parking_lot::{Mutex, RwLock};
use serde_json::Value;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::mpsc;
use super::pending::PendingRequestMap;
@@ -126,13 +127,17 @@ impl CallConnection {
}
};
let (send, recv) = match connection.open_bi().await {
Ok(pair) => pair,
// `open_bi` returns a `BiStream` (ADR-092); split it into halves for
// the call protocol's separate write (request) and read (response)
// pumps. The split is the stdlib idiom; no per-handler wrapper.
let stream = match connection.open_bi().await {
Ok(s) => s,
Err(err) => {
let call_error = CallError::internal(format!("failed to open stream: {err}"));
return ResponseEnvelope::error(request_id, call_error);
}
};
let (recv, send) = tokio::io::split(stream);
let receiver = {
let mut pending = self.pending.lock();
@@ -197,13 +202,16 @@ impl CallConnection {
}
};
let (send, recv) = match connection.open_bi().await {
Ok(pair) => pair,
// `open_bi` returns a `BiStream` (ADR-092); split for the separate
// write (request) and read (subscription events) pumps.
let stream = match connection.open_bi().await {
Ok(s) => s,
Err(err) => {
let call_error = CallError::internal(format!("failed to open stream: {err}"));
return SubscriptionStream::closed(request_id, call_error);
}
};
let (recv, send) = tokio::io::split(stream);
let receiver = {
let mut pending = self.pending.lock();
@@ -235,12 +243,15 @@ impl CallConnection {
self.pending.lock().handle_aborted(request_id);
}
async fn write_request(
async fn write_request<W>(
&self,
send: alknet_core::types::SendStream,
send: W,
request_id: &str,
payload: Value,
) -> Result<(), String> {
) -> Result<(), String>
where
W: AsyncWrite + Unpin,
{
let envelope = EventEnvelope::requested(request_id, payload);
let mut writer = FrameFramedWriter::new(send);
writer
@@ -254,10 +265,13 @@ impl CallConnection {
.connection
.as_ref()
.ok_or_else(|| "no underlying connection (overlay-only)".to_string())?;
let (send, _recv) = connection
// `open_bi` returns a `BiStream` (ADR-092). We only need the write
// half to send the envelope; split and drop the read half.
let stream = connection
.open_bi()
.await
.map_err(|e| format!("failed to open stream: {e}"))?;
let (_recv, send) = tokio::io::split(stream);
let mut writer = FrameFramedWriter::new(send);
writer
.write_frame(envelope)
@@ -266,10 +280,10 @@ impl CallConnection {
}
}
async fn read_stream_until_closed(
recv: alknet_core::types::RecvStream,
pending: &Arc<Mutex<PendingRequestMap>>,
) {
async fn read_stream_until_closed<R>(recv: R, pending: &Arc<Mutex<PendingRequestMap>>)
where
R: AsyncRead + Unpin,
{
let mut reader = FrameFramedReader::new(recv);
while let Ok(envelope) = reader.read_frame().await {
dispatch_envelope(pending, envelope);
@@ -458,17 +472,9 @@ mod tests {
use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
use alknet_core::types::Capabilities;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::time::{Duration, Instant};
fn stub_connection() -> Connection {
Connection::from_stream(
tokio::io::sink(),
tokio::io::empty(),
b"alknet/call".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321)),
)
}
use crate::protocol::sink_empty_connection as stub_connection;
fn external_spec(name: &str) -> OperationSpec {
OperationSpec::new(
crates/alknet-call/src/protocol/dispatch.rs
+19 -28
View File
@@ -285,9 +285,13 @@ impl Dispatcher {
pub(crate) async fn handle_stream(
&self,
connection: Arc<CallConnection>,
send: alknet_core::types::SendStream,
recv: alknet_core::types::RecvStream,
stream: alknet_core::types::BiStream,
) {
// `stream` is a `BiStream` (ADR-092) — `AsyncRead + AsyncWrite + Send
// + Unpin`. Split into the read and write halves the call protocol's
// frame reader/writer consume. The split is the stdlib idiom; no
// per-handler wrapper.
let (recv, send) = tokio::io::split(stream);
let mut reader = FrameFramedReader::new(recv);
let mut writer = FrameFramedWriter::new(send);
@@ -404,11 +408,11 @@ impl Dispatcher {
loop {
match quic.accept_bi().await {
Ok((send, recv)) => {
Ok(stream) => {
let conn = Arc::clone(&connection);
let dispatcher = self.clone();
tokio::spawn(async move {
dispatcher.handle_stream(conn, send, recv).await;
dispatcher.handle_stream(conn, stream).await;
});
}
Err(StreamError::ConnectionClosed) => break,
@@ -458,17 +462,9 @@ mod tests {
use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
use alknet_core::types::Capabilities;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Mutex as StdMutex;
fn stub_connection() -> alknet_core::types::Connection {
alknet_core::types::Connection::from_stream(
tokio::io::sink(),
tokio::io::empty(),
b"alknet/call".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321)),
)
}
use crate::protocol::sink_empty_connection as stub_connection;
struct StaticIdentityProvider {
tokens: StdMutex<HashMap<String, Identity>>,
@@ -1180,10 +1176,9 @@ mod tests {
);
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
let (send, mut sink) = tokio::io::duplex(8 * 1024);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
dp.handle_stream(conn, send, recv).await;
dp.handle_stream(conn, stream).await;
let frames = read_all_frames(&mut sink).await;
assert_eq!(frames.len(), 4, "3 responded + 1 completed");
@@ -1219,10 +1214,9 @@ mod tests {
);
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
let (send, mut sink) = tokio::io::duplex(8 * 1024);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
dp.handle_stream(conn, send, recv).await;
dp.handle_stream(conn, stream).await;
let frames = read_all_frames(&mut sink).await;
assert_eq!(frames.len(), 2, "1 responded + 1 error, no completed");
@@ -1255,10 +1249,9 @@ mod tests {
);
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
let (send, mut sink) = tokio::io::duplex(8 * 1024);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
dp.handle_stream(conn, send, recv).await;
dp.handle_stream(conn, stream).await;
let frames = read_all_frames(&mut sink).await;
assert_eq!(frames.len(), 1, "query: exactly one frame, no completed");
@@ -1286,10 +1279,9 @@ mod tests {
);
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
let (send, mut sink) = tokio::io::duplex(8 * 1024);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
dp.handle_stream(conn, send, recv).await;
dp.handle_stream(conn, stream).await;
let frames = read_all_frames(&mut sink).await;
assert_eq!(frames.len(), 1, "unknown op: single error, no completed");
@@ -1346,13 +1338,12 @@ mod tests {
);
let recv = tokio::io::BufReader::new(std::io::Cursor::new(encode_frame(&request)));
let (send, _sink) = tokio::io::duplex(8 * 1024);
let send = alknet_core::types::SendStream::from_stream(send);
let recv = alknet_core::types::RecvStream::from_stream(recv);
let stream = alknet_core::types::BiStream::from_joined(recv, send);
let conn_clone = Arc::clone(&conn);
let dp_clone = dp.clone();
let handle = tokio::spawn(async move {
dp_clone.handle_stream(conn_clone, send, recv).await;
dp_clone.handle_stream(conn_clone, stream).await;
});
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
crates/alknet-call/src/protocol/mod.rs
+6
View File
@@ -10,3 +10,9 @@ pub mod connection;
pub mod dispatch;
pub mod pending;
pub mod wire;
#[cfg(test)]
mod test_support;
#[cfg(test)]
pub(crate) use test_support::sink_empty_connection;
crates/alknet-call/src/protocol/test_support.rs
+66
View File
@@ -0,0 +1,66 @@
//! Shared test helpers for the call protocol's inline `#[cfg(test)]`
//! modules. Kept here (not in each test module) so the `stub_connection()`
//! shape is defined once — `Connection::from_stream` was removed (ADR-092)
//! and every test stub that previously called it now calls
//! `Connection::from_bidi(SinkEmpty, ...)` via `sink_empty_connection()`.
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::pin::Pin;
use std::task::{Context, Poll};
use alknet_core::types::Connection;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
/// A test-only `AsyncRead + AsyncWrite` pair equivalent to
/// `tokio::io::sink() + tokio::io::empty()`: reads yield EOF immediately
/// (zero bytes), writes discard. Exists because `Connection::from_bidi`
/// (ADR-092 — the only public stream constructor, replacing
/// `from_stream`) requires a single value that implements both traits.
/// Used only to construct a `Connection` for tests that exercise
/// `Connection`-level state (alpn, addr, identity, dispatcher run loop
/// with an immediately-closed accept stream) without ever reading or
/// writing real bytes.
pub(crate) struct SinkEmpty;
impl AsyncRead for SinkEmpty {
fn poll_read(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
_buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
// EOF immediately — mirrors `tokio::io::empty()`.
Poll::Ready(Ok(()))
}
}
impl AsyncWrite for SinkEmpty {
fn poll_write(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
// Discard — mirrors `tokio::io::sink()`.
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
}
/// Construct a `Connection` whose `accept_bi` yields a `SinkEmpty` once,
/// then `ConnectionClosed`. Used by tests that need a `Connection` for
/// `CallConnection::new(conn)` or `adapter.handle(conn, &auth)` without
/// exercising the wire protocol — `SinkEmpty` reads EOF (so the dispatch
/// loop closes immediately) and discards writes.
pub(crate) fn sink_empty_connection() -> Connection {
Connection::from_bidi(
SinkEmpty,
b"alknet/call".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321)),
)
}
crates/alknet-core/src/types.rs
+233 -150
View File
@@ -223,143 +223,193 @@ pub trait ProtocolHandler: Send + Sync + 'static {
async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError>;
}
pub trait BiStream: AsyncRead + AsyncWrite + Send + Unpin {}
// --- BiStream: the handler leaf (ADR-092) ---------------------------------
//
// `accept_bi`/`open_bi` yield `BiStream`, a concrete newtype that boxes the
// joined inner transport. The join happens once in the `BidiStreamSource`
// impl (quinn/iroh via `tokio::io::join`, single-stream via the input
// `AsyncRead + AsyncWrite` boxed directly). The split never crosses a crate
// boundary as part of a constructor: `Connection::from_bidi` is the only
// public stream constructor; `Connection::from_stream` is removed.
enum SendStreamKind {
#[cfg(feature = "quinn")]
Quinn(quinn::SendStream),
#[cfg(feature = "iroh")]
Iroh(iroh::endpoint::SendStream),
Stream(Box<dyn AsyncWrite + Send + Unpin>),
/// Internal helper trait — the union of `AsyncRead + AsyncWrite + Send +
/// Unpin`. Not public; exists only to give `BiStream` a single boxed field.
trait AsyncReadWrite: AsyncRead + AsyncWrite + Send {}
impl<T: AsyncRead + AsyncWrite + Send> AsyncReadWrite for T {}
/// The handler leaf — a bidirectional byte stream (ADR-092).
///
/// `accept_bi`/`open_bi` return a `BiStream`, not a split
/// `(SendStream, RecvStream)` pair. Handlers that want the split halves call
/// `tokio::io::split(&mut *stream)` (the stdlib idiom `tokio::io::split`
/// already provides for `TcpStream` and `TlsStream<TcpStream>`). The
/// split is a stdlib call at the handler boundary, not a per-handler trait
/// wrapper.
///
/// `BiStream: AsyncRead + AsyncWrite + Send + Unpin` by construction. The
/// old `pub trait BiStream: AsyncRead + AsyncWrite + Send + Unpin {}`
/// (ADR-007) is removed — the trait was never consumed, and the concrete
/// struct carries the same trait bounds forward as implied bounds, not a
/// marker trait. The name and the bounds survive; the shape becomes a
/// concrete leaf.
pub struct BiStream {
inner: Box<dyn AsyncReadWrite + Unpin>,
}
enum RecvStreamKind {
#[cfg(feature = "quinn")]
Quinn(quinn::RecvStream),
#[cfg(feature = "iroh")]
Iroh(iroh::endpoint::RecvStream),
Stream(Box<dyn AsyncRead + Send + Unpin>),
impl BiStream {
/// Join a read half and a write half into a single `BiStream`. The join
/// happens once, in the `BidiStreamSource` impl — handlers receive the
/// joined `BiStream` and never see the pair.
///
/// Public so that downstream crates (the channels reassembly path, tests
/// that construct a `BiStream` from independent halves) can join their
/// own halves. The rule this normalizes: **the split never crosses a
/// crate boundary as part of a constructor** — `Connection::from_bidi`
/// takes a joined `BiStream`, and `BiStream::from_joined` is the join.
/// A crate that produces split halves naturally (channels reassembly)
/// joins them itself via this constructor, then hands the `BiStream` to
/// `Connection::from_bidi` (or yields it from its own
/// `BidiStreamSource::accept_bi` impl).
pub fn from_joined<R, W>(reader: R, writer: W) -> Self
where
R: AsyncRead + Send + Unpin + 'static,
W: AsyncWrite + Send + Unpin + 'static,
{
Self {
inner: Box::new(tokio::io::join(reader, writer)),
}
}
/// Wrap a single value that is already `AsyncRead + AsyncWrite` (e.g.
/// `tokio::io::DuplexStream`, `TlsStream<TcpStream>`,
/// `russh::Channel::into_stream()`). Used by the single-stream
/// `BidiStreamSource` impl and by `Connection::from_bidi`.
pub(crate) fn from_bidi<S>(stream: S) -> Self
where
S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
Self {
inner: Box::new(stream),
}
}
}
impl AsyncRead for BiStream {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<io::Result<()>> {
std::pin::Pin::new(self.inner.as_mut()).poll_read(cx, buf)
}
}
impl AsyncWrite for BiStream {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<io::Result<usize>> {
std::pin::Pin::new(self.inner.as_mut()).poll_write(cx, buf)
}
fn poll_flush(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
std::pin::Pin::new(self.inner.as_mut()).poll_flush(cx)
}
fn poll_shutdown(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
std::pin::Pin::new(self.inner.as_mut()).poll_shutdown(cx)
}
}
// --- SendStream / RecvStream: thin newtypes (ADR-092) ---------------------
//
// These remain as the typed-sub-stream leaves for `into_sub_streams()`
// (ADR-074) and the channels reassembly path's `SubStreamHandle` leaves
// (future). They never cross a crate boundary as part of a `Connection`
// constructor — `Connection::from_bidi` is the only public constructor and
// takes a joined `BiStream`. The quinn-welded `SendStreamKind` /
// `RecvStreamKind` enums are gone; the quinn/iroh dispatch moves into the
// `BidiStreamSource` impls (the join happens once, there).
pub struct SendStream {
kind: SendStreamKind,
inner: Box<dyn AsyncWrite + Send + Unpin>,
}
pub struct RecvStream {
kind: RecvStreamKind,
inner: Box<dyn AsyncRead + Send + Unpin>,
}
impl SendStream {
#[cfg(feature = "quinn")]
fn from_quinn(stream: quinn::SendStream) -> Self {
Self {
kind: SendStreamKind::Quinn(stream),
}
}
#[cfg(feature = "iroh")]
fn from_iroh(stream: iroh::endpoint::SendStream) -> Self {
Self {
kind: SendStreamKind::Iroh(stream),
}
}
/// Box a write half into the thin `SendStream` newtype. Used by
/// `into_sub_streams()` (ADR-074) and the channels reassembly path.
/// Not a constructor that feeds `Connection` — the split never crosses
/// a crate boundary as part of a constructor (ADR-092).
pub fn from_stream(stream: impl AsyncWrite + Send + Unpin + 'static) -> Self {
Self {
kind: SendStreamKind::Stream(Box::new(stream)),
inner: Box::new(stream),
}
}
}
impl RecvStream {
#[cfg(feature = "quinn")]
fn from_quinn(stream: quinn::RecvStream) -> Self {
Self {
kind: RecvStreamKind::Quinn(stream),
}
}
#[cfg(feature = "iroh")]
fn from_iroh(stream: iroh::endpoint::RecvStream) -> Self {
Self {
kind: RecvStreamKind::Iroh(stream),
}
}
/// Box a read half into the thin `RecvStream` newtype. Used by
/// `into_sub_streams()` (ADR-074) and the channels reassembly path.
/// Not a constructor that feeds `Connection` — the split never crosses
/// a crate boundary as part of a constructor (ADR-092).
pub fn from_stream(stream: impl AsyncRead + Send + Unpin + 'static) -> Self {
Self {
kind: RecvStreamKind::Stream(Box::new(stream)),
inner: Box::new(stream),
}
}
}
impl AsyncWrite for SendStream {
fn poll_write(
self: std::pin::Pin<&mut Self>,
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<io::Result<usize>> {
match &mut self.get_mut().kind {
#[cfg(feature = "quinn")]
SendStreamKind::Quinn(s) => AsyncWrite::poll_write(std::pin::Pin::new(s), cx, buf),
#[cfg(feature = "iroh")]
SendStreamKind::Iroh(s) => AsyncWrite::poll_write(std::pin::Pin::new(s), cx, buf),
SendStreamKind::Stream(s) => {
AsyncWrite::poll_write(std::pin::Pin::new(s.as_mut()), cx, buf)
}
}
std::pin::Pin::new(self.inner.as_mut()).poll_write(cx, buf)
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
match &mut self.get_mut().kind {
#[cfg(feature = "quinn")]
SendStreamKind::Quinn(s) => AsyncWrite::poll_flush(std::pin::Pin::new(s), cx),
#[cfg(feature = "iroh")]
SendStreamKind::Iroh(s) => AsyncWrite::poll_flush(std::pin::Pin::new(s), cx),
SendStreamKind::Stream(s) => AsyncWrite::poll_flush(std::pin::Pin::new(s.as_mut()), cx),
}
std::pin::Pin::new(self.inner.as_mut()).poll_flush(cx)
}
fn poll_shutdown(
self: std::pin::Pin<&mut Self>,
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
match &mut self.get_mut().kind {
#[cfg(feature = "quinn")]
SendStreamKind::Quinn(s) => AsyncWrite::poll_shutdown(std::pin::Pin::new(s), cx),
#[cfg(feature = "iroh")]
SendStreamKind::Iroh(s) => AsyncWrite::poll_shutdown(std::pin::Pin::new(s), cx),
SendStreamKind::Stream(s) => AsyncWrite::poll_shutdown(std::pin::Pin::new(s), cx),
}
std::pin::Pin::new(self.inner.as_mut()).poll_shutdown(cx)
}
}
impl AsyncRead for RecvStream {
fn poll_read(
self: std::pin::Pin<&mut Self>,
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<io::Result<()>> {
match &mut self.get_mut().kind {
#[cfg(feature = "quinn")]
RecvStreamKind::Quinn(s) => AsyncRead::poll_read(std::pin::Pin::new(s), cx, buf),
#[cfg(feature = "iroh")]
RecvStreamKind::Iroh(s) => AsyncRead::poll_read(std::pin::Pin::new(s), cx, buf),
RecvStreamKind::Stream(s) => {
AsyncRead::poll_read(std::pin::Pin::new(s.as_mut()), cx, buf)
}
}
std::pin::Pin::new(self.inner.as_mut()).poll_read(cx, buf)
}
}
/// Yield bidirectional streams to a `Connection`. Downstream crates implement
/// this trait to add connection shapes (channels, a future transport, a test
/// double beyond the `from_stream` case) without editing `alknet-core`. See
/// double beyond the single-stream case) without editing `alknet-core`. See
/// ADR-070 for the full rationale and ADR-065 for the yield-once contract the
/// `StreamBidiStreamSource` impl preserves.
/// `StreamBidiStreamSource` impl preserves. The return type is `BiStream`
/// (ADR-092) — the join happens once, in the impl, not per-handler.
#[async_trait]
pub trait BidiStreamSource: Send + Sync + 'static {
/// Yield the next bidirectional stream this connection provides.
@@ -372,14 +422,14 @@ pub trait BidiStreamSource: Send + Sync + 'static {
/// `ConnectionClosed` on all subsequent calls.
/// - Channels: yields one bidi stream per channel, `ConnectionClosed`
/// when the channels connection closes.
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError>;
async fn accept_bi(&self) -> Result<BiStream, StreamError>;
/// Open a bidirectional stream to the peer.
///
/// Single-stream sources return `StreamClosed` (a single stream cannot
/// open new application streams — ADR-065). QUIC and channels sources
/// open new streams.
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError>;
async fn open_bi(&self) -> Result<BiStream, StreamError>;
/// The peer's address, if available. Informational (NAT/proxy).
fn remote_addr(&self) -> Option<SocketAddr>;
@@ -401,22 +451,22 @@ struct QuinnBidiStreamSource {
#[cfg(feature = "quinn")]
#[async_trait]
impl BidiStreamSource for QuinnBidiStreamSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn accept_bi(&self) -> Result<BiStream, StreamError> {
let (send, recv) = self
.conn
.accept_bi()
.await
.map_err(map_quinn_connection_error)?;
Ok((SendStream::from_quinn(send), RecvStream::from_quinn(recv)))
Ok(BiStream::from_joined(recv, send))
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn open_bi(&self) -> Result<BiStream, StreamError> {
let (send, recv) = self
.conn
.open_bi()
.await
.map_err(map_quinn_connection_error)?;
Ok((SendStream::from_quinn(send), RecvStream::from_quinn(recv)))
Ok(BiStream::from_joined(recv, send))
}
fn remote_addr(&self) -> Option<SocketAddr> {
@@ -439,22 +489,22 @@ struct IrohBidiStreamSource {
#[cfg(feature = "iroh")]
#[async_trait]
impl BidiStreamSource for IrohBidiStreamSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn accept_bi(&self) -> Result<BiStream, StreamError> {
let (send, recv) = self
.conn
.accept_bi()
.await
.map_err(map_iroh_connection_error)?;
Ok((SendStream::from_iroh(send), RecvStream::from_iroh(recv)))
Ok(BiStream::from_joined(recv, send))
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn open_bi(&self) -> Result<BiStream, StreamError> {
let (send, recv) = self
.conn
.open_bi()
.await
.map_err(map_iroh_connection_error)?;
Ok((SendStream::from_iroh(send), RecvStream::from_iroh(recv)))
Ok(BiStream::from_joined(recv, send))
}
fn remote_addr(&self) -> Option<SocketAddr> {
@@ -469,25 +519,25 @@ impl BidiStreamSource for IrohBidiStreamSource {
/// Single-stream `BidiStreamSource` (TCP+TLS, SSH channel, WebTransport
/// stream, wasm stream — ADR-065). Crate-private; constructed via
/// `Connection::from_stream` / `from_bidi` (no feature gate). `accept_bi`
/// yields the underlying stream once, then `ConnectionClosed`; `open_bi`
/// returns `StreamClosed`.
/// `Connection::from_bidi` (no feature gate). `accept_bi` yields the
/// underlying `BiStream` once, then `ConnectionClosed`; `open_bi` returns
/// `StreamClosed`.
struct StreamBidiStreamSource {
stream: Mutex<Option<(SendStream, RecvStream)>>,
stream: Mutex<Option<BiStream>>,
remote_addr: Option<SocketAddr>,
}
#[async_trait]
impl BidiStreamSource for StreamBidiStreamSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn accept_bi(&self) -> Result<BiStream, StreamError> {
let mut guard = self.stream.lock().expect("stream mutex poisoned");
match guard.take() {
Some(pair) => Ok(pair),
Some(stream) => Ok(stream),
None => Err(StreamError::ConnectionClosed),
}
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn open_bi(&self) -> Result<BiStream, StreamError> {
Err(StreamError::StreamClosed)
}
@@ -535,37 +585,30 @@ impl Connection {
}
}
/// Construct a `Connection` from a pre-split read/write pair.
/// `accept_bi()` yields this pair once, then returns `ConnectionClosed`.
/// `open_bi()` returns `StreamClosed` (a single stream can't open new streams).
pub fn from_stream(
send: impl AsyncWrite + Send + Unpin + 'static,
recv: impl AsyncRead + Send + Unpin + 'static,
alpn: Vec<u8>,
remote_addr: Option<SocketAddr>,
) -> Self {
Self {
source: Box::new(StreamBidiStreamSource {
stream: Mutex::new(Some((
SendStream::from_stream(send),
RecvStream::from_stream(recv),
))),
remote_addr,
}),
alpn,
identity: OnceLock::new(),
}
}
/// Convenience for a single bidirectional stream (e.g. `TlsStream<TcpStream>`).
/// Splits internally via `tokio::io::split`.
/// Construct a `Connection` from a single bidirectional stream (e.g.
/// `tokio::io::DuplexStream`, `TlsStream<TcpStream>`,
/// `russh::Channel::into_stream()`). The stream is wrapped in a
/// `BiStream` (ADR-092) and yielded by `accept_bi` once, then
/// `ConnectionClosed`. `open_bi` returns `StreamClosed` (a single
/// stream can't open new application streams — ADR-065).
///
/// This is the only public stream constructor (ADR-092): the split
/// never crosses a crate boundary as part of a constructor. Handlers
/// that want the split halves call `tokio::io::split(&mut *stream)` on
/// the `BiStream` they receive from `accept_bi`.
pub fn from_bidi(
stream: impl AsyncRead + AsyncWrite + Send + Unpin + 'static,
alpn: Vec<u8>,
remote_addr: Option<SocketAddr>,
) -> Self {
let (recv, send) = tokio::io::split(stream);
Self::from_stream(send, recv, alpn, remote_addr)
Self {
source: Box::new(StreamBidiStreamSource {
stream: Mutex::new(Some(BiStream::from_bidi(stream))),
remote_addr,
}),
alpn,
identity: OnceLock::new(),
}
}
/// Construct from a caller-supplied `BidiStreamSource` impl. The
@@ -591,12 +634,14 @@ impl Connection {
///
/// Handlers that loop `accept_bi` (e.g. `TtyAdapter`) get one session
/// per single-stream connection; handlers that call once (e.g.
/// `HttpAdapter`) get the stream directly. Both are correct.
pub async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
/// `HttpAdapter`) get the stream directly. Both are correct. The
/// return type is `BiStream` (ADR-092); handlers that want the split
/// halves call `tokio::io::split` on the `BiStream`.
pub async fn accept_bi(&self) -> Result<BiStream, StreamError> {
self.source.accept_bi().await
}
pub async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
pub async fn open_bi(&self) -> Result<BiStream, StreamError> {
self.source.open_bi().await
}
@@ -657,21 +702,21 @@ mod from_source_tests {
/// 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<Option<(SendStream, RecvStream)>>,
stream: Mutex<Option<BiStream>>,
addr: Option<SocketAddr>,
closed: Arc<Mutex<Option<(u32, String)>>>,
}
#[async_trait]
impl BidiStreamSource for RecordingSource {
async fn accept_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn accept_bi(&self) -> Result<BiStream, StreamError> {
match self.stream.lock().expect("mock mutex poisoned").take() {
Some(pair) => Ok(pair),
Some(stream) => Ok(stream),
None => Err(StreamError::ConnectionClosed),
}
}
async fn open_bi(&self) -> Result<(SendStream, RecvStream), StreamError> {
async fn open_bi(&self) -> Result<BiStream, StreamError> {
Err(StreamError::StreamClosed)
}
@@ -693,19 +738,16 @@ mod from_source_tests {
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.
// One duplex: the mock holds end `a`; the test driver holds end `b`
// (split into send_b/recv_b) to echo back. The mock's `accept_bi`
// yields end `a` as a `BiStream`; the driver reads/writes end `b`.
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),
))),
stream: Mutex::new(Some(BiStream::from_bidi(a))),
addr,
closed: Arc::clone(&recorded),
},
@@ -718,22 +760,22 @@ mod from_source_tests {
// 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");
// accept_bi delegates to RecordingSource::accept_bi and yields a BiStream.
let mut stream = 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");
// Write via the mock's BiStream -> arrives at the driver's recv_b.
stream.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.
// Driver writes back -> arrives at the mock's BiStream.
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");
stream.read_exact(&mut buf).await.expect("read round-trips");
assert_eq!(&buf, b"world");
// Second accept_bi delegates to RecordingSource::accept_bi -> ConnectionClosed.
@@ -763,11 +805,52 @@ mod from_source_tests {
mod tests {
use super::*;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::pin::Pin;
use std::task::{Context, Poll};
/// A test-only `AsyncRead + AsyncWrite` pair equivalent to
/// `tokio::io::sink()` + `tokio::io::empty()`: reads yield EOF
/// immediately (zero bytes), writes discard. Exists because
/// `Connection::from_bidi` requires a single value that implements
/// both traits (ADR-092 — the split-pair `from_stream` constructor is
/// removed). Used only to construct a `Connection` for tests that
/// exercise `Connection`-level state (alpn, addr, identity) without
/// ever reading or writing the stream.
struct SinkEmpty;
impl AsyncRead for SinkEmpty {
fn poll_read(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
_buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
// EOF immediately — mirrors `tokio::io::empty()`.
Poll::Ready(Ok(()))
}
}
impl AsyncWrite for SinkEmpty {
fn poll_write(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
// Discard — mirrors `tokio::io::sink()`.
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
fn test_connection() -> Connection {
Connection::from_stream(
tokio::io::sink(),
tokio::io::empty(),
Connection::from_bidi(
SinkEmpty,
b"alknet/test".to_vec(),
Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1234)),
)
@@ -863,7 +946,7 @@ mod tests {
}
#[test]
fn connection_remote_alpn_and_addr_from_stream() {
fn connection_remote_alpn_and_addr_from_bidi() {
let conn = test_connection();
assert_eq!(conn.remote_alpn(), b"alknet/test");
assert_eq!(
crates/alknet-http/src/server/adapter.rs
+30 -141
View File
@@ -8,9 +8,7 @@
//! from `gateway_routes`; `/openapi.json` serves the `to_openapi` projection
//! of the registry.
use std::io;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use async_trait::async_trait;
@@ -229,12 +227,14 @@ impl ProtocolHandler for HttpAdapter {
let _ = connection.set_identity(identity);
}
let (send, recv) = connection
// `accept_bi` returns a `BiStream` (ADR-092) — already
// `AsyncRead + AsyncWrite + Send + Unpin`. No wrapper needed; pass
// it directly to `serve_io` via `TokioIo::new`.
let stream = connection
.accept_bi()
.await
.map_err(stream_error_to_handler)?;
let io = QuicStream::new(send, recv);
self.serve_io(io).await
self.serve_io(stream).await
}
}
@@ -268,51 +268,6 @@ fn stream_error_to_handler(e: StreamError) -> HandlerError {
HandlerError::from(e)
}
struct QuicStream {
send: alknet_core::types::SendStream,
recv: alknet_core::types::RecvStream,
}
impl QuicStream {
fn new(send: alknet_core::types::SendStream, recv: alknet_core::types::RecvStream) -> Self {
Self { send, recv }
}
}
impl AsyncRead for QuicStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<io::Result<()>> {
Pin::new(&mut self.recv).poll_read(cx, buf)
}
}
impl AsyncWrite for QuicStream {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<io::Result<usize>> {
Pin::new(&mut self.send).poll_write(cx, buf)
}
fn poll_flush(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
Pin::new(&mut self.send).poll_flush(cx)
}
fn poll_shutdown(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
Pin::new(&mut self.send).poll_shutdown(cx)
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -418,29 +373,26 @@ mod tests {
async fn send_request_and_read_response(
request: &[u8],
) -> (String, tokio::task::JoinHandle<()>) {
let (mut client_send, server_recv) = duplex(8 * 1024);
let (server_send, mut client_recv) = duplex(8 * 1024);
let server_io = QuicStreamDuplex {
read: server_recv,
write: server_send,
};
// One duplex: `server_io` is the server's end (passed to `serve_io`);
// `client_io` is the client's end (writes requests, reads responses).
// `tokio::io::duplex` yields two `DuplexStream`s, each
// `AsyncRead + AsyncWrite + Send + Unpin` — the same bounds `BiStream`
// exposes (ADR-092), so no wrapper is needed.
let (server_io, mut client_io) = duplex(8 * 1024);
let adapter = HttpAdapter::new(provider(), empty_registry());
let handle = tokio::spawn(async move {
adapter.serve_io(server_io).await.ok();
});
client_send.write_all(request).await.unwrap();
client_send.flush().await.unwrap();
client_io.write_all(request).await.unwrap();
client_io.flush().await.unwrap();
let mut response = Vec::new();
let mut buf = [0u8; 4096];
loop {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
client_recv.read(&mut buf),
)
.await
match tokio::time::timeout(std::time::Duration::from_secs(5), client_io.read(&mut buf))
.await
{
Ok(Ok(0)) => break,
Ok(Ok(n)) => response.extend_from_slice(&buf[..n]),
@@ -453,45 +405,6 @@ mod tests {
(response_str, handle)
}
struct QuicStreamDuplex {
read: tokio::io::DuplexStream,
write: tokio::io::DuplexStream,
}
impl AsyncRead for QuicStreamDuplex {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<io::Result<()>> {
Pin::new(&mut self.read).poll_read(cx, buf)
}
}
impl AsyncWrite for QuicStreamDuplex {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<io::Result<usize>> {
Pin::new(&mut self.write).poll_write(cx, buf)
}
fn poll_flush(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
Pin::new(&mut self.write).poll_flush(cx)
}
fn poll_shutdown(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<io::Result<()>> {
Pin::new(&mut self.write).poll_shutdown(cx)
}
}
#[tokio::test]
async fn handle_serves_http_request_over_mock_quic_stream() {
let request = b"GET /healthz HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
@@ -509,29 +422,21 @@ mod tests {
let extra = Router::new().route("/v1/foo", get(|| async { (StatusCode::OK, "foo-body") }));
let adapter = HttpAdapter::new(provider(), empty_registry()).with_extra_routes(extra);
let (mut client_send, server_recv) = duplex(8 * 1024);
let (server_send, mut client_recv) = duplex(8 * 1024);
let server_io = QuicStreamDuplex {
read: server_recv,
write: server_send,
};
let (server_io, mut client_io) = duplex(8 * 1024);
let handle = tokio::spawn(async move {
adapter.serve_io(server_io).await.ok();
});
let request = b"GET /v1/foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
client_send.write_all(request).await.unwrap();
client_send.flush().await.unwrap();
client_io.write_all(request).await.unwrap();
client_io.flush().await.unwrap();
let mut response = Vec::new();
let mut buf = [0u8; 4096];
loop {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
client_recv.read(&mut buf),
)
.await
match tokio::time::timeout(std::time::Duration::from_secs(5), client_io.read(&mut buf))
.await
{
Ok(Ok(0)) => break,
Ok(Ok(n)) => response.extend_from_slice(&buf[..n]),
@@ -556,29 +461,21 @@ mod tests {
);
let adapter = HttpAdapter::new(provider(), empty_registry()).with_extra_routes(extra);
let (mut client_send, server_recv) = duplex(8 * 1024);
let (server_send, mut client_recv) = duplex(8 * 1024);
let server_io = QuicStreamDuplex {
read: server_recv,
write: server_send,
};
let (server_io, mut client_io) = duplex(8 * 1024);
let handle = tokio::spawn(async move {
adapter.serve_io(server_io).await.ok();
});
let request = b"GET /healthz HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
client_send.write_all(request).await.unwrap();
client_send.flush().await.unwrap();
client_io.write_all(request).await.unwrap();
client_io.flush().await.unwrap();
let mut response = Vec::new();
let mut buf = [0u8; 4096];
loop {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
client_recv.read(&mut buf),
)
.await
match tokio::time::timeout(std::time::Duration::from_secs(5), client_io.read(&mut buf))
.await
{
Ok(Ok(0)) => break,
Ok(Ok(n)) => response.extend_from_slice(&buf[..n]),
@@ -597,25 +494,17 @@ mod tests {
}
async fn serve_and_read(adapter: HttpAdapter, request: &[u8]) -> String {
let (mut client_send, server_recv) = duplex(8 * 1024);
let (server_send, mut client_recv) = duplex(8 * 1024);
let server_io = QuicStreamDuplex {
read: server_recv,
write: server_send,
};
let (server_io, mut client_io) = duplex(8 * 1024);
let handle = tokio::spawn(async move {
adapter.serve_io(server_io).await.ok();
});
client_send.write_all(request).await.unwrap();
client_send.flush().await.unwrap();
client_io.write_all(request).await.unwrap();
client_io.flush().await.unwrap();
let mut response = Vec::new();
let mut buf = [0u8; 4096];
loop {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
client_recv.read(&mut buf),
)
.await
match tokio::time::timeout(std::time::Duration::from_secs(5), client_io.read(&mut buf))
.await
{
Ok(Ok(0)) => break,
Ok(Ok(n)) => response.extend_from_slice(&buf[..n]),
crates/alknet-tty/src/adapter.rs
+14 -7
View File
@@ -118,8 +118,8 @@ impl ProtocolHandler for TtyAdapter {
let _ = connection.set_identity(identity);
}
loop {
let (send, recv) = match connection.accept_bi().await {
Ok(pair) => pair,
let stream = match connection.accept_bi().await {
Ok(stream) => stream,
Err(StreamError::ConnectionClosed) => break,
Err(StreamError::StreamClosed) => break,
Err(e) => return Err(HandlerError::from(e)),
@@ -128,7 +128,14 @@ impl ProtocolHandler for TtyAdapter {
let ownership = self.ownership.clone();
let identity = auth.identity.clone();
tokio::spawn(async move {
let _ = drive_session(send, recv, backends, ownership, identity).await;
// `stream` is a `BiStream` (ADR-092) — `AsyncRead + AsyncWrite
// + Send + Unpin`. Split into halves for `drive_session`
// (which takes separate `AsyncWrite` + `AsyncRead` args). The
// split is the stdlib idiom for `TcpStream`-style duplex
// streams; no per-handler wrapper.
let (client_read, client_write) = tokio::io::split(stream);
let _ =
drive_session(client_write, client_read, backends, ownership, identity).await;
});
}
Ok(())
@@ -164,10 +171,10 @@ async fn send_negotiation_error<W: AsyncWrite + Unpin>(
/// Drive a `alknet/tty` session end-to-end over a bidi stream.
///
/// `client_send` / `client_recv` are the two halves of the bidi stream (QUIC
/// `SendStream` / `RecvStream`). Returns when the session is complete (exit
/// chunk sent, stream closed) or when the stream is reset (cancel-cleanup
/// path — no exit chunk sent).
/// `client_send` / `client_recv` are the two halves of the bidi stream
/// (split from the `BiStream` yielded by `accept_bi` via `tokio::io::split`).
/// Returns when the session is complete (exit chunk sent, stream closed) or
/// when the stream is reset (cancel-cleanup path — no exit chunk sent).
///
/// This is the per-stream session driver — the counterpart to the POC's
/// `session::drive_session` (`/workspace/alknet-tty-poc/src/session.rs`),