From 42d7884fb1fc751c83e274db2c7ef89959f670fb Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Tue, 1 Sep 2026 21:09:29 +0000 Subject: [PATCH 1/5] add p3 feature --- Cargo.toml | 70 ++++++++++++++++++++++++++++++++++++++++++++++--- axum/Cargo.toml | 19 +++++++++++++- axum/src/lib.rs | 1 + src/lib.rs | 40 +++++++++++++++++++++------- 4 files changed, 116 insertions(+), 14 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 386afa8..3b4a3b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,9 +13,13 @@ repository.workspace = true rust-version.workspace = true [features] -default = ["json"] +default = ["json", "p2"] json = ["dep:serde", "dep:serde_json"] +# WASI backend selection. Exactly one of these must be enabled. +p2 = ["dep:wasip2"] +p3 = ["dep:wasip3"] + [dependencies] anyhow.workspace = true async-task.workspace = true @@ -27,9 +31,11 @@ http.workspace = true itoa.workspace = true pin-project-lite.workspace = true slab.workspace = true -wasip2.workspace = true wstd-macro.workspace = true +wasip2 = { workspace = true, optional = true } +wasip3 = { workspace = true, optional = true } + # optional serde = { workspace = true, optional = true } serde_json = { workspace = true, optional = true } @@ -43,6 +49,62 @@ humantime.workspace = true serde = { workspace = true, features = ["derive"] } serde_json.workspace = true +[[test]] +name = "http_first_byte_timeout" +required-features = ["p2"] + +[[test]] +name = "http_get" +required-features = ["p2"] + +[[test]] +name = "http_get_json" +required-features = ["p2"] + +[[test]] +name = "http_handle_error_code" +required-features = ["p2"] + +[[test]] +name = "http_post" +required-features = ["p2"] + +[[test]] +name = "http_post_json" +required-features = ["p2"] + +[[test]] +name = "http_timeout" +required-features = ["p2"] + +[[test]] +name = "sleep" +required-features = ["p2"] + +[[example]] +name = "complex_http_client" +required-features = ["p2"] + +[[example]] +name = "http_client" +required-features = ["p2"] + +[[example]] +name = "http_server" +required-features = ["p2"] + +[[example]] +name = "http_server_proxy" +required-features = ["p2"] + +[[example]] +name = "tcp_echo_server" +required-features = ["p2"] + +[[example]] +name = "tcp_stream_client" +required-features = ["p2"] + [workspace] members = [ "axum", @@ -95,13 +157,13 @@ test-programs = { path = "test-programs" } tower-service = "0.3.3" ureq = { version = "3.1", default-features = false, features = ["json"] } wasip2 = "1.0" -wstd = { path = ".", version = "=0.6.8" } +wasip3 = "0.8" +wstd = { path = ".", version = "=0.6.8", default-features = false } wstd-axum = { path = "./axum", version = "=0.6.8" } wstd-axum-macro = { path = "./axum/macro", version = "=0.6.8" } wstd-macro = { path = "./macro", version = "=0.6.8" } [package.metadata.docs.rs] -all-features = true targets = [ "wasm32-wasip2" ] diff --git a/axum/Cargo.toml b/axum/Cargo.toml index 154a021..baa5da4 100644 --- a/axum/Cargo.toml +++ b/axum/Cargo.toml @@ -13,12 +13,29 @@ rust-version.workspace = true [dependencies] axum.workspace = true tower-service.workspace = true -wstd.workspace = true +wstd = { workspace = true, features = ["json"] } wstd-axum-macro.workspace = true +[features] +default = ["p2"] +p2 = ["wstd/p2"] +p3 = ["wstd/p3"] + [dev-dependencies] anyhow.workspace = true futures-concurrency.workspace = true serde = { workspace = true, features = ["derive"] } serde_qs.workspace = true axum = { workspace = true, features = ["query", "json", "macros"] } + +[[example]] +name = "hello_world" +required-features = ["p2"] + +[[example]] +name = "hello_world_nomacro" +required-features = ["p2"] + +[[example]] +name = "weather" +required-features = ["p2"] diff --git a/axum/src/lib.rs b/axum/src/lib.rs index 3272b91..9536c89 100644 --- a/axum/src/lib.rs +++ b/axum/src/lib.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "p2")] //! Support for the [`axum`] web server framework in wasi-http components, via //! [`wstd`]. //! diff --git a/src/lib.rs b/src/lib.rs index ebc673d..0dc8ed7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,7 +18,7 @@ //! **TCP echo server** //! //! ```rust,no_run -#![doc = include_str!("../examples/tcp_echo_server.rs")] +#![cfg_attr(feature = "p2", doc = include_str!("../examples/tcp_echo_server.rs"))] //! ``` //! //! **HTTP Client** @@ -30,7 +30,7 @@ //! **HTTP Server** //! //! ```rust,no_run -#![doc = include_str!("../examples/http_server.rs")] +#![cfg_attr(feature = "p2", doc = include_str!("../examples/http_server.rs"))] //! ``` //! //! # Design Decisions @@ -55,30 +55,52 @@ //! These are unique capabilities provided by WASI 0.2, and because this library //! is specific to that are exposed from here. +// Exactly one WASI backend must be selected. See the `p2`/`p3` features. +#[cfg(all(feature = "p2", feature = "p3"))] +compile_error!( + "the `p2` and `p3` features are mutually exclusive — enable exactly one WASI backend" +); +#[cfg(not(any(feature = "p2", feature = "p3")))] +compile_error!("exactly one of the `p2` or `p3` features must be enabled"); + +#[cfg(feature = "p2")] pub mod future; +#[cfg(feature = "p2")] #[macro_use] pub mod http; +#[cfg(feature = "p2")] pub mod io; pub mod iter; +#[cfg(feature = "p2")] pub mod net; +#[cfg(feature = "p2")] pub mod rand; +#[cfg(feature = "p2")] pub mod runtime; +#[cfg(feature = "p2")] pub mod task; +#[cfg(feature = "p2")] pub mod time; -pub use wstd_macro::attr_macro_http_server as http_server; -pub use wstd_macro::attr_macro_main as main; -pub use wstd_macro::attr_macro_test as test; +#[cfg(feature = "p2")] +pub use wstd_macro::{ + attr_macro_http_server as http_server, attr_macro_main as main, attr_macro_test as test, +}; -// Re-export the wasip2 crate for use only by `wstd-macro` macros. The proc -// macros need to generate code that uses these definitions, but we don't want -// to treat it as part of our public API with regards to semver, so we keep it -// under `__internal` as well as doc(hidden) to indicate it is private. +// Re-export the active WASI backend crate for use only by `wstd-macro` macros. +// The proc macros need to generate code that uses these definitions, but we +// don't want to treat it as part of our public API with regards to semver, so +// we keep it under `__internal` as well as doc(hidden) to indicate it is +// private. #[doc(hidden)] pub mod __internal { + #[cfg(feature = "p2")] pub use wasip2; + #[cfg(feature = "p3")] + pub use wasip3; } +#[cfg(feature = "p2")] pub mod prelude { pub use crate::future::FutureExt as _; pub use crate::io::AsyncRead as _; From 4a6e65cc71b452d39000f998e47c04e605c38dd0 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 2 Sep 2026 12:24:53 +0000 Subject: [PATCH 2/5] add p3 test CI job --- .github/workflows/ci.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 1362609..9681c60 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -54,6 +54,9 @@ jobs: - name: wstd tests run: cargo test -p wstd -p wstd-axum --target wasm32-wasip2 -- --nocapture + - name: p3 wstd tests + run: cargo test -p wstd -p wstd-axum --target wasm32-wasip2 --no-default-features --features p3 -- --nocapture + - name: test-programs tests run: cargo test -p test-programs -- --nocapture if: steps.creds.outcome == 'success' From ea32edf32998964ef18ab766043b98a226fbf5f4 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 3 Sep 2026 14:00:43 +0000 Subject: [PATCH 3/5] use cfg alias --- Cargo.toml | 4 ++++ axum/Cargo.toml | 3 +++ axum/build.rs | 10 ++++++++++ axum/src/lib.rs | 2 +- build.rs | 10 ++++++++++ src/lib.rs | 32 ++++++++++++++++---------------- 6 files changed, 44 insertions(+), 17 deletions(-) create mode 100644 axum/build.rs create mode 100644 build.rs diff --git a/Cargo.toml b/Cargo.toml index 3b4a3b9..90a90cd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,9 @@ json = ["dep:serde", "dep:serde_json"] p2 = ["dep:wasip2"] p3 = ["dep:wasip3"] +[build-dependencies] +cfg_aliases.workspace = true + [dependencies] anyhow.workspace = true async-task.workspace = true @@ -134,6 +137,7 @@ async-task = "4.7" axum = { version = "0.8.6", default-features = false } bytes = "1.10.1" cargo_metadata = "0.22" +cfg_aliases = "0.2" clap = { version = "4.5.26", features = ["derive"] } futures-core = "0.3.19" futures-lite = "1.12.0" diff --git a/axum/Cargo.toml b/axum/Cargo.toml index baa5da4..a27ce78 100644 --- a/axum/Cargo.toml +++ b/axum/Cargo.toml @@ -21,6 +21,9 @@ default = ["p2"] p2 = ["wstd/p2"] p3 = ["wstd/p3"] +[build-dependencies] +cfg_aliases.workspace = true + [dev-dependencies] anyhow.workspace = true futures-concurrency.workspace = true diff --git a/axum/build.rs b/axum/build.rs new file mode 100644 index 0000000..5a5f457 --- /dev/null +++ b/axum/build.rs @@ -0,0 +1,10 @@ +use cfg_aliases::cfg_aliases; + +fn main() { + cfg_aliases! { + // TODO https://github.com/bytecodealliance/wstd/issues/147: Swap these + // to use `target_env` instead. + p2: { feature = "p2" }, + p3: { feature = "p3" }, + } +} diff --git a/axum/src/lib.rs b/axum/src/lib.rs index 9536c89..c730fd2 100644 --- a/axum/src/lib.rs +++ b/axum/src/lib.rs @@ -1,4 +1,4 @@ -#![cfg(feature = "p2")] +#![cfg(p2)] //! Support for the [`axum`] web server framework in wasi-http components, via //! [`wstd`]. //! diff --git a/build.rs b/build.rs new file mode 100644 index 0000000..5a5f457 --- /dev/null +++ b/build.rs @@ -0,0 +1,10 @@ +use cfg_aliases::cfg_aliases; + +fn main() { + cfg_aliases! { + // TODO https://github.com/bytecodealliance/wstd/issues/147: Swap these + // to use `target_env` instead. + p2: { feature = "p2" }, + p3: { feature = "p3" }, + } +} diff --git a/src/lib.rs b/src/lib.rs index 0dc8ed7..64ac672 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,7 +18,7 @@ //! **TCP echo server** //! //! ```rust,no_run -#![cfg_attr(feature = "p2", doc = include_str!("../examples/tcp_echo_server.rs"))] +#![cfg_attr(p2, doc = include_str!("../examples/tcp_echo_server.rs"))] //! ``` //! //! **HTTP Client** @@ -30,7 +30,7 @@ //! **HTTP Server** //! //! ```rust,no_run -#![cfg_attr(feature = "p2", doc = include_str!("../examples/http_server.rs"))] +#![cfg_attr(p2, doc = include_str!("../examples/http_server.rs"))] //! ``` //! //! # Design Decisions @@ -56,33 +56,33 @@ //! is specific to that are exposed from here. // Exactly one WASI backend must be selected. See the `p2`/`p3` features. -#[cfg(all(feature = "p2", feature = "p3"))] +#[cfg(all(p2, p3))] compile_error!( "the `p2` and `p3` features are mutually exclusive — enable exactly one WASI backend" ); -#[cfg(not(any(feature = "p2", feature = "p3")))] +#[cfg(not(any(p2, p3)))] compile_error!("exactly one of the `p2` or `p3` features must be enabled"); -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod future; -#[cfg(feature = "p2")] +#[cfg(p2)] #[macro_use] pub mod http; -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod io; pub mod iter; -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod net; -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod rand; -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod runtime; -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod task; -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod time; -#[cfg(feature = "p2")] +#[cfg(p2)] pub use wstd_macro::{ attr_macro_http_server as http_server, attr_macro_main as main, attr_macro_test as test, }; @@ -94,13 +94,13 @@ pub use wstd_macro::{ // private. #[doc(hidden)] pub mod __internal { - #[cfg(feature = "p2")] + #[cfg(p2)] pub use wasip2; - #[cfg(feature = "p3")] + #[cfg(p3)] pub use wasip3; } -#[cfg(feature = "p2")] +#[cfg(p2)] pub mod prelude { pub use crate::future::FutureExt as _; pub use crate::io::AsyncRead as _; From 0487755bbbe73dc9bf0ce5ac6dada4d3d3175eb1 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 2 Sep 2026 14:23:48 +0000 Subject: [PATCH 4/5] Require `&mut` for Streams --- examples/tcp_echo_server.rs | 6 ++- src/http/body.rs | 10 ++--- src/io/read.rs | 4 +- src/io/stdio.rs | 8 ++-- src/io/streams.rs | 35 ++++++++--------- src/io/write.rs | 4 +- src/net/tcp_listener.rs | 4 +- src/net/tcp_stream.rs | 77 ++++++++++++++++--------------------- 8 files changed, 68 insertions(+), 80 deletions(-) diff --git a/examples/tcp_echo_server.rs b/examples/tcp_echo_server.rs index f1dd895..3b02009 100644 --- a/examples/tcp_echo_server.rs +++ b/examples/tcp_echo_server.rs @@ -4,7 +4,7 @@ use wstd::net::TcpListener; #[wstd::main] async fn main() -> io::Result<()> { - let listener = TcpListener::bind("127.0.0.1:8080").await?; + let mut listener = TcpListener::bind("127.0.0.1:8080").await?; println!("Listening on {}", listener.local_addr()?); println!("type `nc localhost 8080` to create a TCP client"); @@ -14,7 +14,9 @@ async fn main() -> io::Result<()> { println!("Accepted from: {}", stream.peer_addr()?); wstd::runtime::spawn(async move { // If echo copy fails, we can ignore it. - let _ = io::copy(&stream, &stream).await; + let mut stream = stream; + let (mut read_half, mut write_half) = stream.split(); + let _ = io::copy(&mut read_half, &mut write_half).await; }) .detach(); } diff --git a/src/http/body.rs b/src/http/body.rs index 95e8a2e..e04e624 100644 --- a/src/http/body.rs +++ b/src/http/body.rs @@ -77,7 +77,7 @@ impl Body { match self.0 { BodyInner::Incoming(incoming) => incoming.send(outgoing_body).await, BodyInner::Boxed(box_body) => { - let out_stream = AsyncOutputStream::new( + let mut out_stream = AsyncOutputStream::new( outgoing_body .write() .expect("outgoing body already written"), @@ -108,7 +108,7 @@ impl Body { } } BodyInner::Complete { data, trailers } => { - let out_stream = AsyncOutputStream::new( + let mut out_stream = AsyncOutputStream::new( outgoing_body .write() .expect("outgoing body already written"), @@ -348,14 +348,14 @@ impl Incoming { } async fn send(self, outgoing_body: WasiOutgoingBody) -> Result<(), Error> { let in_body = self.body; - let in_stream = + let mut in_stream = AsyncInputStream::new(in_body.stream().expect("incoming body already read")); - let out_stream = AsyncOutputStream::new( + let mut out_stream = AsyncOutputStream::new( outgoing_body .write() .expect("outgoing body already written"), ); - in_stream.copy_to(&out_stream).await.map_err(|e| { + in_stream.copy_to(&mut out_stream).await.map_err(|e| { Error::from(e).context("copying incoming body stream to outgoing body stream") })?; drop(in_stream); diff --git a/src/io/read.rs b/src/io/read.rs index a6a95da..474188e 100644 --- a/src/io/read.rs +++ b/src/io/read.rs @@ -28,7 +28,7 @@ pub trait AsyncRead { // If the `AsyncRead` implementation is an unbuffered wrapper around an // `AsyncInputStream`, some I/O operations can be more efficient. #[inline] - fn as_async_input_stream(&self) -> Option<&io::AsyncInputStream> { + fn as_async_input_stream(&mut self) -> Option<&mut io::AsyncInputStream> { None } } @@ -45,7 +45,7 @@ impl AsyncRead for &mut R { } #[inline] - fn as_async_input_stream(&self) -> Option<&io::AsyncInputStream> { + fn as_async_input_stream(&mut self) -> Option<&mut io::AsyncInputStream> { (**self).as_async_input_stream() } } diff --git a/src/io/stdio.rs b/src/io/stdio.rs index b2ac153..e403986 100644 --- a/src/io/stdio.rs +++ b/src/io/stdio.rs @@ -43,8 +43,8 @@ impl AsyncRead for Stdin { } #[inline] - fn as_async_input_stream(&self) -> Option<&AsyncInputStream> { - Some(&self.stream) + fn as_async_input_stream(&mut self) -> Option<&mut AsyncInputStream> { + Some(&mut self.stream) } } @@ -93,7 +93,7 @@ impl AsyncWrite for Stdout { } #[inline] - fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> { + fn as_async_output_stream(&mut self) -> Option<&mut AsyncOutputStream> { self.stream.as_async_output_stream() } } @@ -143,7 +143,7 @@ impl AsyncWrite for Stderr { } #[inline] - fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> { + fn as_async_output_stream(&mut self) -> Option<&mut AsyncOutputStream> { self.stream.as_async_output_stream() } } diff --git a/src/io/streams.rs b/src/io/streams.rs index 3676d21..8644d66 100644 --- a/src/io/streams.rs +++ b/src/io/streams.rs @@ -26,7 +26,7 @@ impl AsyncInputStream { stream, } } - fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<()> { + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<()> { // Lazily initialize the AsyncPollable let subscription = self .subscription @@ -43,12 +43,11 @@ impl AsyncInputStream { } } /// Await for read readiness. - async fn ready(&self) { + async fn ready(&mut self) { poll_fn(|cx| self.poll_ready(cx)).await } /// Asynchronously read from the input stream. - /// This method is the same as [`AsyncRead::read`], but doesn't require a `&mut self`. - pub async fn read(&self, buf: &mut [u8]) -> std::io::Result { + pub async fn read(&mut self, buf: &mut [u8]) -> std::io::Result { let read = loop { self.ready().await; // Ideally, the ABI would be able to read directly into buf. @@ -76,7 +75,7 @@ impl AsyncInputStream { /// Move the entire contents of an input stream directly into an output /// stream, until the input stream has closed. This operation is optimized /// to avoid copying stream contents into and out of memory. - pub async fn copy_to(&self, writer: &AsyncOutputStream) -> std::io::Result { + pub async fn copy_to(&mut self, writer: &mut AsyncOutputStream) -> std::io::Result { let mut written = 0; loop { self.ready().await; @@ -128,7 +127,7 @@ impl AsyncRead for AsyncInputStream { } #[inline] - fn as_async_input_stream(&self) -> Option<&AsyncInputStream> { + fn as_async_input_stream(&mut self) -> Option<&mut AsyncInputStream> { Some(self) } } @@ -150,9 +149,10 @@ impl AsyncInputChunkStream { impl futures_lite::stream::Stream for AsyncInputChunkStream { type Item = Result, std::io::Error>; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match self.stream.poll_ready(cx) { + let this = self.get_mut(); + match this.stream.poll_ready(cx) { Poll::Pending => Poll::Pending, - Poll::Ready(()) => match self.stream.stream.read(self.chunk_size as u64) { + Poll::Ready(()) => match this.stream.stream.read(this.chunk_size as u64) { Ok(r) if r.is_empty() => Poll::Pending, Ok(r) => Poll::Ready(Some(Ok(r))), Err(StreamError::LastOperationFailed(err)) => { @@ -233,7 +233,7 @@ impl AsyncOutputStream { } } /// Await write readiness. - async fn ready(&self) { + async fn ready(&mut self) { // Lazily initialize the AsyncPollable let subscription = self .subscription @@ -241,15 +241,14 @@ impl AsyncOutputStream { // Wait on readiness subscription.wait_for().await; } - /// Asynchronously write to the output stream. This method is the same as - /// [`AsyncWrite::write`], but doesn't require a `&mut self`. + /// Asynchronously write to the output stream. /// /// Awaits for write readiness, and then performs at most one write to the /// output stream. Returns how much of the argument `buf` was written, or /// a `std::io::Error` indicating either an error returned by the stream write /// using the debug string provided by the WASI error, or else that the, /// indicated by `std::io::ErrorKind::ConnectionReset`. - pub async fn write(&self, buf: &[u8]) -> std::io::Result { + pub async fn write(&mut self, buf: &[u8]) -> std::io::Result { // Loops at most twice. loop { match self.stream.check_write() { @@ -280,9 +279,8 @@ impl AsyncOutputStream { } } - /// Asynchronously write to the output stream. This method is the same as - /// [`AsyncWrite::write_all`], but doesn't require a `&mut self`. - pub async fn write_all(&self, buf: &[u8]) -> std::io::Result<()> { + /// Asynchronously write to the output stream. + pub async fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> { let mut to_write = &buf[0..]; loop { let bytes_written = self.write(to_write).await?; @@ -297,14 +295,11 @@ impl AsyncOutputStream { /// awaits until the flush is complete and the output stream is ready for /// writing again. /// - /// This method is the same as [`AsyncWrite::flush`], but doesn't require - /// a `&mut self`. - /// /// Fails with a `std::io::Error` indicating either an error returned by /// the stream flush, using the debug string provided by the WASI error, /// or else that the stream is closed, indicated by /// `std::io::ErrorKind::ConnectionReset`. - pub async fn flush(&self) -> std::io::Result<()> { + pub async fn flush(&mut self) -> std::io::Result<()> { match self.stream.flush() { Ok(()) => { self.ready().await; @@ -330,7 +325,7 @@ impl AsyncWrite for AsyncOutputStream { } #[inline] - fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> { + fn as_async_output_stream(&mut self) -> Option<&mut AsyncOutputStream> { Some(self) } } diff --git a/src/io/write.rs b/src/io/write.rs index 79cf0d9..12fd6ae 100644 --- a/src/io/write.rs +++ b/src/io/write.rs @@ -20,7 +20,7 @@ pub trait AsyncWrite { // If the `AsyncWrite` implementation is an unbuffered wrapper around an // `AsyncOutputStream`, some I/O operations can be more efficient. #[inline] - fn as_async_output_stream(&self) -> Option<&io::AsyncOutputStream> { + fn as_async_output_stream(&mut self) -> Option<&mut io::AsyncOutputStream> { None } } @@ -42,7 +42,7 @@ impl AsyncWrite for &mut W { } #[inline] - fn as_async_output_stream(&self) -> Option<&io::AsyncOutputStream> { + fn as_async_output_stream(&mut self) -> Option<&mut io::AsyncOutputStream> { (**self).as_async_output_stream() } } diff --git a/src/net/tcp_listener.rs b/src/net/tcp_listener.rs index 9a1f57a..b7af12b 100644 --- a/src/net/tcp_listener.rs +++ b/src/net/tcp_listener.rs @@ -56,7 +56,7 @@ impl TcpListener { } /// Returns an iterator over the connections being received on this listener. - pub fn incoming(&self) -> Incoming<'_> { + pub fn incoming(&mut self) -> Incoming<'_> { Incoming { listener: self } } } @@ -64,7 +64,7 @@ impl TcpListener { /// An iterator that infinitely accepts connections on a TcpListener. #[derive(Debug)] pub struct Incoming<'a> { - listener: &'a TcpListener, + listener: &'a mut TcpListener, } impl<'a> AsyncIterator for Incoming<'a> { diff --git a/src/net/tcp_stream.rs b/src/net/tcp_stream.rs index af3674a..977fb29 100644 --- a/src/net/tcp_stream.rs +++ b/src/net/tcp_stream.rs @@ -86,8 +86,17 @@ impl TcpStream { Ok(format!("{addr:?}")) } - pub fn split(&self) -> (ReadHalf<'_>, WriteHalf<'_>) { - (ReadHalf(self), WriteHalf(self)) + pub fn split(&mut self) -> (ReadHalf<'_>, WriteHalf<'_>) { + ( + ReadHalf { + stream: &mut self.input, + socket: &self.socket, + }, + WriteHalf { + stream: &mut self.output, + socket: &self.socket, + }, + ) } } @@ -104,18 +113,8 @@ impl io::AsyncRead for TcpStream { self.input.read(buf).await } - fn as_async_input_stream(&self) -> Option<&AsyncInputStream> { - Some(&self.input) - } -} - -impl io::AsyncRead for &TcpStream { - async fn read(&mut self, buf: &mut [u8]) -> io::Result { - self.input.read(buf).await - } - - fn as_async_input_stream(&self) -> Option<&AsyncInputStream> { - (**self).as_async_input_stream() + fn as_async_input_stream(&mut self) -> Option<&mut AsyncInputStream> { + Some(&mut self.input) } } @@ -128,64 +127,56 @@ impl io::AsyncWrite for TcpStream { self.output.flush().await } - fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> { - Some(&self.output) + fn as_async_output_stream(&mut self) -> Option<&mut AsyncOutputStream> { + Some(&mut self.output) } } -impl io::AsyncWrite for &TcpStream { - async fn write(&mut self, buf: &[u8]) -> io::Result { - self.output.write(buf).await - } - - async fn flush(&mut self) -> io::Result<()> { - self.output.flush().await - } +pub struct ReadHalf<'a> { + stream: &'a mut AsyncInputStream, + socket: &'a TcpSocket, +} - fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> { - (**self).as_async_output_stream() +impl<'a> Drop for ReadHalf<'a> { + fn drop(&mut self) { + let _ = self + .socket + .shutdown(wasip2::sockets::tcp::ShutdownType::Receive); } } -pub struct ReadHalf<'a>(&'a TcpStream); impl<'a> io::AsyncRead for ReadHalf<'a> { async fn read(&mut self, buf: &mut [u8]) -> io::Result { - self.0.read(buf).await + self.stream.read(buf).await } - fn as_async_input_stream(&self) -> Option<&AsyncInputStream> { - self.0.as_async_input_stream() + fn as_async_input_stream(&mut self) -> Option<&mut AsyncInputStream> { + self.stream.as_async_input_stream() } } -impl<'a> Drop for ReadHalf<'a> { - fn drop(&mut self) { - let _ = self - .0 - .socket - .shutdown(wasip2::sockets::tcp::ShutdownType::Receive); - } +pub struct WriteHalf<'a> { + stream: &'a mut AsyncOutputStream, + socket: &'a TcpSocket, } -pub struct WriteHalf<'a>(&'a TcpStream); impl<'a> io::AsyncWrite for WriteHalf<'a> { async fn write(&mut self, buf: &[u8]) -> io::Result { - self.0.write(buf).await + self.stream.write(buf).await } async fn flush(&mut self) -> io::Result<()> { - self.0.flush().await + self.stream.flush().await } - fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> { - self.0.as_async_output_stream() + fn as_async_output_stream(&mut self) -> Option<&mut AsyncOutputStream> { + self.stream.as_async_output_stream() } } impl<'a> Drop for WriteHalf<'a> { fn drop(&mut self) { let _ = self - .0 .socket .shutdown(wasip2::sockets::tcp::ShutdownType::Send); } From 443ef72ef26593fd0da98c1bcfe0424eb72768ae Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 2 Sep 2026 15:09:22 +0000 Subject: [PATCH 5/5] remove &self stream methods --- src/http/body.rs | 2 +- src/io/streams.rs | 95 ++++++++++++++--------------------------------- 2 files changed, 29 insertions(+), 68 deletions(-) diff --git a/src/http/body.rs b/src/http/body.rs index e04e624..b4de743 100644 --- a/src/http/body.rs +++ b/src/http/body.rs @@ -3,7 +3,7 @@ use crate::http::{ error::Context as _, fields::{header_map_from_wasi, header_map_to_wasi}, }; -use crate::io::{AsyncInputStream, AsyncOutputStream}; +use crate::io::{AsyncInputStream, AsyncOutputStream, AsyncWrite}; use crate::runtime::{AsyncPollable, Reactor, WaitFor}; pub use ::http_body::{Body as HttpBody, Frame, SizeHint}; diff --git a/src/io/streams.rs b/src/io/streams.rs index 8644d66..34fd3b0 100644 --- a/src/io/streams.rs +++ b/src/io/streams.rs @@ -46,32 +46,6 @@ impl AsyncInputStream { async fn ready(&mut self) { poll_fn(|cx| self.poll_ready(cx)).await } - /// Asynchronously read from the input stream. - pub async fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - let read = loop { - self.ready().await; - // Ideally, the ABI would be able to read directly into buf. - // However, with the default generated bindings, it returns a - // newly allocated vec, which we need to copy into buf. - match self.stream.read(buf.len() as u64) { - // A read of 0 bytes from WASI's `read` doesn't mean - // end-of-stream as it does in Rust. However, `self.ready()` - // cannot guarantee that at least one byte is ready for - // reading, so in this case we try again. - Ok(r) if r.is_empty() => continue, - Ok(r) => break r, - // 0 bytes from Rust's `read` means end-of-stream. - Err(StreamError::Closed) => return Ok(0), - Err(StreamError::LastOperationFailed(err)) => { - return Err(std::io::Error::other(err.to_debug_string())); - } - } - }; - let len = read.len(); - buf[0..len].copy_from_slice(&read); - Ok(len) - } - /// Move the entire contents of an input stream directly into an output /// stream, until the input stream has closed. This operation is optimized /// to avoid copying stream contents into and out of memory. @@ -123,7 +97,28 @@ impl AsyncInputStream { impl AsyncRead for AsyncInputStream { async fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - Self::read(self, buf).await + let read = loop { + self.ready().await; + // Ideally, the ABI would be able to read directly into buf. + // However, with the default generated bindings, it returns a + // newly allocated vec, which we need to copy into buf. + match self.stream.read(buf.len() as u64) { + // A read of 0 bytes from WASI's `read` doesn't mean + // end-of-stream as it does in Rust. However, `self.ready()` + // cannot guarantee that at least one byte is ready for + // reading, so in this case we try again. + Ok(r) if r.is_empty() => continue, + Ok(r) => break r, + // 0 bytes from Rust's `read` means end-of-stream. + Err(StreamError::Closed) => return Ok(0), + Err(StreamError::LastOperationFailed(err)) => { + return Err(std::io::Error::other(err.to_debug_string())); + } + } + }; + let len = read.len(); + buf[0..len].copy_from_slice(&read); + Ok(len) } #[inline] @@ -241,14 +236,11 @@ impl AsyncOutputStream { // Wait on readiness subscription.wait_for().await; } - /// Asynchronously write to the output stream. - /// - /// Awaits for write readiness, and then performs at most one write to the - /// output stream. Returns how much of the argument `buf` was written, or - /// a `std::io::Error` indicating either an error returned by the stream write - /// using the debug string provided by the WASI error, or else that the, - /// indicated by `std::io::ErrorKind::ConnectionReset`. - pub async fn write(&mut self, buf: &[u8]) -> std::io::Result { +} + +impl AsyncWrite for AsyncOutputStream { + // Required methods + async fn write(&mut self, buf: &[u8]) -> std::io::Result { // Loops at most twice. loop { match self.stream.check_write() { @@ -278,28 +270,7 @@ impl AsyncOutputStream { } } } - - /// Asynchronously write to the output stream. - pub async fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> { - let mut to_write = &buf[0..]; - loop { - let bytes_written = self.write(to_write).await?; - to_write = &to_write[bytes_written..]; - if to_write.is_empty() { - return Ok(()); - } - } - } - - /// Asyncronously flush the output stream. Initiates a flush, and then - /// awaits until the flush is complete and the output stream is ready for - /// writing again. - /// - /// Fails with a `std::io::Error` indicating either an error returned by - /// the stream flush, using the debug string provided by the WASI error, - /// or else that the stream is closed, indicated by - /// `std::io::ErrorKind::ConnectionReset`. - pub async fn flush(&mut self) -> std::io::Result<()> { + async fn flush(&mut self) -> std::io::Result<()> { match self.stream.flush() { Ok(()) => { self.ready().await; @@ -313,16 +284,6 @@ impl AsyncOutputStream { } } } -} - -impl AsyncWrite for AsyncOutputStream { - // Required methods - async fn write(&mut self, buf: &[u8]) -> std::io::Result { - Self::write(self, buf).await - } - async fn flush(&mut self) -> std::io::Result<()> { - Self::flush(self).await - } #[inline] fn as_async_output_stream(&mut self) -> Option<&mut AsyncOutputStream> {