diff --git a/Cargo.toml b/Cargo.toml index b2e2627..7ee77cd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,7 @@ serde_json = { workspace = true, optional = true } wasip2.workspace = true [target.'cfg(all(target_os = "wasi", target_env = "p3"))'.dependencies] -wasip3 = { workspace = true, features = ["async-spawn"] } +wasip3 = { workspace = true, features = ["async-spawn", "http-compat"] } [dev-dependencies] anyhow.workspace = true diff --git a/axum/examples/hello_world.rs b/axum/examples/hello_world.rs index 9931344..90a9cbd 100644 --- a/axum/examples/hello_world.rs +++ b/axum/examples/hello_world.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - //! Run with //! //! ```sh diff --git a/axum/examples/hello_world_nomacro.rs b/axum/examples/hello_world_nomacro.rs index bc0a39b..f3068d1 100644 --- a/axum/examples/hello_world_nomacro.rs +++ b/axum/examples/hello_world_nomacro.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - //! Run with //! //! ```sh diff --git a/axum/examples/weather.rs b/axum/examples/weather.rs index 0df8f97..5163fcc 100644 --- a/axum/examples/weather.rs +++ b/axum/examples/weather.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - //! This demo app shows a Axum based wasi-http server making an arbitrary //! number of http requests as part of serving a single response. //! diff --git a/axum/src/lib.rs b/axum/src/lib.rs index c5a2af1..3272b91 100644 --- a/axum/src/lib.rs +++ b/axum/src/lib.rs @@ -1,4 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] //! Support for the [`axum`] web server framework in wasi-http components, via //! [`wstd`]. //! diff --git a/examples/complex_http_client.rs b/examples/complex_http_client.rs index 7c3e470..703e931 100644 --- a/examples/complex_http_client.rs +++ b/examples/complex_http_client.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use anyhow::{Result, anyhow}; use clap::{ArgAction, Parser}; use std::str::FromStr; diff --git a/examples/http_client.rs b/examples/http_client.rs index 178fbeb..f4465a8 100644 --- a/examples/http_client.rs +++ b/examples/http_client.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use anyhow::{Result, anyhow}; use clap::{ArgAction, Parser}; use wstd::http::{Body, BodyExt, Client, Method, Request, Uri}; diff --git a/examples/http_server.rs b/examples/http_server.rs index fa67518..8449f83 100644 --- a/examples/http_server.rs +++ b/examples/http_server.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use anyhow::{Context, Result}; use futures_lite::stream::{once_future, unfold}; use http_body_util::{BodyExt, StreamBody}; diff --git a/examples/http_server_proxy.rs b/examples/http_server_proxy.rs index 8d0cf01..ccad608 100644 --- a/examples/http_server_proxy.rs +++ b/examples/http_server_proxy.rs @@ -1,6 +1,3 @@ -#![cfg_attr(not(all(target_os = "wasi", target_env = "p2")), no_main)] -#![cfg(all(target_os = "wasi", target_env = "p2"))] - //! Run the example with: //! ```sh //! cargo build --example http_server_proxy --target=wasm32-wasip2 diff --git a/macro/src/lib.rs b/macro/src/lib.rs index 6498377..96ba632 100644 --- a/macro/src/lib.rs +++ b/macro/src/lib.rs @@ -121,8 +121,10 @@ pub fn attr_macro_http_server(_attr: TokenStream, item: TokenStream) -> TokenStr } quote! { + #[cfg(target_env = "p2")] struct TheServer; + #[cfg(target_env = "p2")] impl ::wstd::__internal::wasip2::exports::http::incoming_handler::Guest for TheServer { fn handle( request: ::wstd::__internal::wasip2::http::types::IncomingRequest, @@ -146,8 +148,39 @@ pub fn attr_macro_http_server(_attr: TokenStream, item: TokenStream) -> TokenStr } } + #[cfg(target_env = "p2")] ::wstd::__internal::wasip2::http::proxy::export!(TheServer with_types_in ::wstd::__internal::wasip2); + #[cfg(target_env = "p3")] + struct TheServer; + + #[cfg(target_env = "p3")] + impl ::wstd::__internal::wasip3::exports::http::handler::Guest for TheServer { + async fn handle( + request: ::wstd::__internal::wasip3::http::types::Request, + ) -> ::core::result::Result< + ::wstd::__internal::wasip3::http::types::Response, + ::wstd::__internal::wasip3::http::types::ErrorCode, + > { + #(#attrs)* + #vis #run_async fn __run(#inputs) #output { + #body + } + + let request = ::wstd::http::request::try_from_incoming(request) + .map_err(::wstd::http::response::error_code)?; + let response = __run(request) #run_await + .map_err(::wstd::http::response::error_code)?; + ::wstd::http::response::try_into_outgoing(response) + .map_err(::wstd::http::response::error_code) + } + } + + #[cfg(target_env = "p3")] + ::wstd::__internal::wasip3::http::service::export!( + TheServer with_types_in ::wstd::__internal::wasip3 + ); + // Provide an actual function named `main`. // // WASI HTTP server components don't use a traditional `main` function. diff --git a/src/http/body.rs b/src/http/body/sys/p2.rs similarity index 100% rename from src/http/body.rs rename to src/http/body/sys/p2.rs diff --git a/src/http/body/sys/p3.rs b/src/http/body/sys/p3.rs new file mode 100644 index 0000000..6805f8f --- /dev/null +++ b/src/http/body/sys/p3.rs @@ -0,0 +1,286 @@ +use crate::http::{Error, HeaderMap, error::Context as _}; + +pub use ::http_body::{Body as HttpBody, Frame, SizeHint}; +pub use bytes::Bytes; + +use http::header::CONTENT_LENGTH; +use http_body_util::{BodyExt, combinators::UnsyncBoxBody}; +use std::fmt; +use std::pin::Pin; +use std::task::{Context, Poll}; + +pub mod util { + pub use http_body_util::*; +} + +/// An HTTP body. +/// +/// Bodies can be constructed from bytes, strings, streams, or any +/// [`http_body::Body`], and incoming WASI HTTP bodies use the same public type. +#[derive(Debug)] +pub struct Body(BodyInner); + +#[derive(Debug)] +enum BodyInner { + Boxed(UnsyncBoxBody), + Incoming { + body: UnsyncBoxBody, + size_hint: BodyHint, + }, + Complete { + data: Bytes, + trailers: Option, + }, +} + +impl Body { + /// Convert this body into an `http_body::Body` trait object. + pub fn into_boxed_body(self) -> UnsyncBoxBody { + fn map_e(_: std::convert::Infallible) -> Error { + unreachable!() + } + + match self.0 { + BodyInner::Complete { data, trailers } => http_body_util::Full::new(data) + .map_err(map_e) + .with_trailers(async move { Ok(trailers).transpose() }) + .boxed_unsync(), + BodyInner::Boxed(body) => body, + BodyInner::Incoming { body, .. } => body, + } + } + + /// Collect the entire contents of this body into memory. + pub async fn contents(&mut self) -> Result<&[u8], Error> { + match &mut self.0 { + BodyInner::Complete { data, .. } => Ok(data), + inner => { + let previous = std::mem::replace( + inner, + BodyInner::Complete { + data: Bytes::new(), + trailers: None, + }, + ); + let body = match previous { + BodyInner::Boxed(body) | BodyInner::Incoming { body, .. } => body, + BodyInner::Complete { .. } => unreachable!(), + }; + let collected = body.collect().await?; + let trailers = collected.trailers().cloned(); + *inner = BodyInner::Complete { + data: collected.to_bytes(), + trailers, + }; + let BodyInner::Complete { data, .. } = inner else { + unreachable!() + }; + Ok(data) + } + } + } + + /// Collect the entire contents of this body as [`Bytes`]. + pub async fn bytes_contents(&mut self) -> Result { + self.contents().await?; + let BodyInner::Complete { data, .. } = &self.0 else { + unreachable!() + }; + Ok(data.clone()) + } + + /// Return the body length when it is known. + pub fn content_length(&self) -> Option { + match &self.0 { + BodyInner::Boxed(body) => body.size_hint().exact(), + BodyInner::Incoming { size_hint, .. } => size_hint.content_length(), + BodyInner::Complete { data, .. } => Some(data.len() as u64), + } + } + + /// Construct an empty body. + pub fn empty() -> Self { + Self(BodyInner::Complete { + data: Bytes::new(), + trailers: None, + }) + } + + /// Collect the entire contents of this body as UTF-8. + pub async fn str_contents(&mut self) -> Result<&str, Error> { + std::str::from_utf8(self.contents().await?).context("decoding body contents as string") + } + + /// Construct a body by serializing a value as JSON. + #[cfg(feature = "json")] + pub fn from_json(data: &T) -> Result { + Ok(Self::from(serde_json::to_vec(data)?)) + } + + /// Collect and deserialize this body as JSON. + #[cfg(feature = "json")] + pub async fn json serde::Deserialize<'a>>(&mut self) -> Result { + serde_json::from_str(self.str_contents().await?).context("decoding body contents as json") + } + + pub(crate) fn from_incoming( + body: wasip3::http_compat::IncomingBody, + size_hint: BodyHint, + ) -> Self + where + T: wasip3::http_compat::IncomingMessage + Send + 'static, + { + Self(BodyInner::Incoming { + body: body.map_err(Into::into).boxed_unsync(), + size_hint, + }) + } + + /// Construct a body from a stream of byte chunks. + pub fn from_stream(stream: S) -> Self + where + S: futures_lite::Stream + Send + 'static, + S::Item: Into, + { + use futures_lite::StreamExt; + Self::from_http_body(http_body_util::StreamBody::new( + stream.map(|bytes| Ok::<_, Error>(Frame::data(bytes.into()))), + )) + } + + /// Construct a body from a fallible stream of byte chunks. + pub fn from_try_stream(stream: S) -> Self + where + S: futures_lite::Stream> + Send + 'static, + D: Into, + E: std::error::Error + Send + Sync + 'static, + { + use futures_lite::StreamExt; + Self::from_http_body(http_body_util::StreamBody::new( + stream.map(|bytes| Ok::<_, Error>(Frame::data(bytes?.into()))), + )) + } + + /// Construct a body from an [`http_body::Body`]. + pub fn from_http_body(body: B) -> Self + where + B: HttpBody + Send + 'static, + B::Data: Into, + B::Error: Into, + { + Self(BodyInner::Boxed( + body.map_frame(|frame| frame.map_data(Into::into)) + .map_err(Into::into) + .boxed_unsync(), + )) + } +} + +impl HttpBody for Body { + type Data = Bytes; + type Error = Error; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + match &mut self.get_mut().0 { + BodyInner::Boxed(body) => Pin::new(body).poll_frame(cx), + BodyInner::Incoming { body, .. } => Pin::new(body).poll_frame(cx), + BodyInner::Complete { data, trailers } => { + if !data.is_empty() { + Poll::Ready(Some(Ok(Frame::data(std::mem::take(data))))) + } else if let Some(trailers) = trailers.take() { + Poll::Ready(Some(Ok(Frame::trailers(trailers)))) + } else { + Poll::Ready(None) + } + } + } + } +} + +impl From<()> for Body { + fn from(_: ()) -> Self { + Self::empty() + } +} +impl From<&[u8]> for Body { + fn from(bytes: &[u8]) -> Self { + Self::from(bytes.to_owned()) + } +} +impl From> for Body { + fn from(bytes: Vec) -> Self { + Self::from(Bytes::from(bytes)) + } +} +impl From for Body { + fn from(data: Bytes) -> Self { + Self(BodyInner::Complete { + data, + trailers: None, + }) + } +} +impl From<&str> for Body { + fn from(data: &str) -> Self { + Self::from(data.as_bytes()) + } +} +impl From for Body { + fn from(data: String) -> Self { + Self::from(data.into_bytes()) + } +} + +impl From for Body { + fn from(stream: crate::io::AsyncInputStream) -> Self { + use futures_lite::StreamExt; + Self::from_http_body(http_body_util::StreamBody::new(stream.into_stream().map( + |result| { + result + .map(|bytes| Frame::data(Bytes::from(bytes))) + .map_err(Error::from) + }, + ))) + } +} + +#[derive(Clone, Copy, Debug)] +pub enum BodyHint { + ContentLength(u64), + Unknown, +} + +impl BodyHint { + pub fn from_headers(headers: &HeaderMap) -> Result { + match headers.get(CONTENT_LENGTH) { + Some(value) => Ok(Self::ContentLength( + std::str::from_utf8(value.as_ref()) + .map_err(|_| InvalidContentLength)? + .parse() + .map_err(|_| InvalidContentLength)?, + )), + None => Ok(Self::Unknown), + } + } + + fn content_length(self) -> Option { + match self { + Self::ContentLength(length) => Some(length), + Self::Unknown => None, + } + } +} + +#[derive(Debug)] +pub struct InvalidContentLength; + +impl fmt::Display for InvalidContentLength { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Invalid Content-Length header") + } +} + +impl std::error::Error for InvalidContentLength {} diff --git a/src/http/client.rs b/src/http/client.rs index 3676fa8..8ffab1d 100644 --- a/src/http/client.rs +++ b/src/http/client.rs @@ -1,9 +1,13 @@ use super::{Body, Error, Request, Response}; use crate::http::request::try_into_outgoing; use crate::http::response::try_from_incoming; +#[cfg(target_env = "p2")] use crate::io::AsyncPollable; use crate::time::Duration; +#[cfg(target_env = "p2")] use wasip2::http::types::RequestOptions as WasiRequestOptions; +#[cfg(target_env = "p3")] +use wasip3::http::types::RequestOptions as WasiRequestOptions; /// An HTTP client. #[derive(Debug, Clone)] @@ -25,31 +29,46 @@ impl Client { /// Send an HTTP request. pub async fn send>(&self, req: Request) -> Result, Error> { - let (wasi_req, body) = try_into_outgoing(req)?; - let body = body.into(); - let wasi_body = wasi_req.body().unwrap(); - - // 1. Start sending the request head - let res = wasip2::http::outgoing_handler::handle(wasi_req, self.wasi_options()?)?; - - let ((), body) = futures_lite::future::try_zip( - async move { - // 3. send the body: - body.send(wasi_body).await - }, - async move { - // 4. Receive the response - AsyncPollable::new(res.subscribe()).wait_for().await; - - // NOTE: the first `unwrap` is to ensure readiness, the second `unwrap` - // is to trap if we try and get the response more than once. The final - // `?` is to raise the actual error if there is one. - let res = res.get().unwrap().unwrap()?; - try_from_incoming(res) - }, - ) - .await?; - Ok(body) + #[cfg(target_env = "p3")] + { + let mut request = req.map(Into::into); + if let Some(options) = self.wasi_options()? { + request + .extensions_mut() + .insert(wasip3::http_compat::RequestOptionsExtension(options)); + } + let request = try_into_outgoing(request)?; + try_from_incoming(wasip3::http::client::send(request).await?) + } + + #[cfg(target_env = "p2")] + { + let (wasi_req, body) = try_into_outgoing(req)?; + let body = body.into(); + let wasi_body = wasi_req.body().unwrap(); + + // 1. Start sending the request head + let res = wasip2::http::outgoing_handler::handle(wasi_req, self.wasi_options()?)?; + + let ((), body) = futures_lite::future::try_zip( + async move { + // 3. send the body: + body.send(wasi_body).await + }, + async move { + // 4. Receive the response + AsyncPollable::new(res.subscribe()).wait_for().await; + + // NOTE: the first `unwrap` is to ensure readiness, the second `unwrap` + // is to trap if we try and get the response more than once. The final + // `?` is to raise the actual error if there is one. + let res = res.get().unwrap().unwrap()?; + try_from_incoming(res) + }, + ) + .await?; + Ok(body) + } } /// Set timeout on connecting to HTTP server @@ -92,6 +111,7 @@ struct RequestOptions { between_bytes_timeout: Option, } +#[cfg(target_env = "p2")] impl RequestOptions { fn to_wasi(&self) -> Result { let wasi = WasiRequestOptions::new(); @@ -120,3 +140,20 @@ impl RequestOptions { Ok(wasi) } } + +#[cfg(target_env = "p3")] +impl RequestOptions { + fn to_wasi(&self) -> Result { + let wasi = WasiRequestOptions::new(); + if let Some(timeout) = self.connect_timeout { + wasi.set_connect_timeout(Some(timeout.0))?; + } + if let Some(timeout) = self.first_byte_timeout { + wasi.set_first_byte_timeout(Some(timeout.0))?; + } + if let Some(timeout) = self.between_bytes_timeout { + wasi.set_between_bytes_timeout(Some(timeout.0))?; + } + Ok(wasi) + } +} diff --git a/src/http/error.rs b/src/http/error.rs index a4f22b0..96a2634 100644 --- a/src/http/error.rs +++ b/src/http/error.rs @@ -6,7 +6,10 @@ pub use crate::http::body::InvalidContentLength; pub use anyhow::Context; pub use http::header::{InvalidHeaderName, InvalidHeaderValue}; pub use http::method::InvalidMethod; +#[cfg(target_env = "p2")] pub use wasip2::http::types::{ErrorCode, HeaderError}; +#[cfg(target_env = "p3")] +pub use wasip3::http::types::{ErrorCode, HeaderError}; pub type Error = anyhow::Error; /// The `http` result type. diff --git a/src/http/fields.rs b/src/http/fields.rs index de6df16..a1a820f 100644 --- a/src/http/fields.rs +++ b/src/http/fields.rs @@ -1,11 +1,13 @@ -pub use http::header::{HeaderMap, HeaderName, HeaderValue}; +pub(super) use http::header::{HeaderMap, HeaderName, HeaderValue}; use super::{Error, error::Context}; use wasip2::http::types::Fields; pub(crate) fn header_map_from_wasi(wasi_fields: Fields) -> Result { let mut output = HeaderMap::new(); - for (key, value) in wasi_fields.entries() { + let entries = wasi_fields.entries(); + + for (key, value) in entries { let key = HeaderName::from_bytes(key.as_bytes()).with_context(|| format!("header name {key}"))?; let value = diff --git a/src/http/method.rs b/src/http/method.rs index d1882a8..f702ef1 100644 --- a/src/http/method.rs +++ b/src/http/method.rs @@ -1,7 +1,6 @@ -use wasip2::http::types::Method as WasiMethod; - -pub use http::Method; +use http::Method; use http::method::InvalidMethod; +use wasip2::http::types::Method as WasiMethod; pub(crate) fn to_wasi_method(value: Method) -> WasiMethod { match value { diff --git a/src/http/mod.rs b/src/http/mod.rs index 39f0a40..e2c160e 100644 --- a/src/http/mod.rs +++ b/src/http/mod.rs @@ -1,25 +1,63 @@ //! HTTP networking support //! +pub use http::Method; +pub use http::header::{HeaderMap, HeaderName, HeaderValue}; pub use http::status::StatusCode; -pub use http::uri::{Authority, PathAndQuery, Uri}; +pub use http::uri::{Authority, InvalidUri, PathAndQuery, Scheme, Uri}; #[doc(inline)] pub use body::{Body, util::BodyExt}; pub use client::Client; pub use error::{Error, ErrorCode, Result}; -pub use fields::{HeaderMap, HeaderName, HeaderValue}; -pub use method::Method; pub use request::Request; pub use response::Response; -pub use scheme::{InvalidUri, Scheme}; -pub mod body; +pub mod body { + mod sys { + #[cfg(target_env = "p2")] + pub(super) mod p2; + #[cfg(target_env = "p3")] + pub(super) mod p3; + } + #[cfg(target_env = "p2")] + pub use sys::p2::*; + #[cfg(target_env = "p3")] + pub use sys::p3::*; +} + +pub mod request { + mod sys { + #[cfg(target_env = "p2")] + pub(super) mod p2; + #[cfg(target_env = "p3")] + pub(super) mod p3; + } + #[cfg(target_env = "p2")] + pub use sys::p2::*; + #[cfg(target_env = "p3")] + pub use sys::p3::*; +} + +pub mod response { + mod sys { + #[cfg(target_env = "p2")] + pub(super) mod p2; + #[cfg(target_env = "p3")] + pub(super) mod p3; + } + #[cfg(target_env = "p2")] + pub use sys::p2::*; + #[cfg(target_env = "p3")] + pub use sys::p3::*; +} mod client; pub mod error; +#[cfg(target_env = "p2")] mod fields; +#[cfg(target_env = "p2")] mod method; -pub mod request; -pub mod response; +#[cfg(target_env = "p2")] mod scheme; +#[cfg(target_env = "p2")] pub mod server; diff --git a/src/http/request.rs b/src/http/request/sys/p2.rs similarity index 99% rename from src/http/request.rs rename to src/http/request/sys/p2.rs index 6694d03..9558a3e 100644 --- a/src/http/request.rs +++ b/src/http/request/sys/p2.rs @@ -1,4 +1,4 @@ -use super::{ +use crate::http::{ Authority, HeaderMap, PathAndQuery, Uri, body::{Body, BodyHint}, error::{Context, Error, ErrorCode}, @@ -12,7 +12,6 @@ use wasip2::http::types::IncomingRequest; pub use http::request::{Builder, Request}; // TODO: go back and add json stuff??? - pub(crate) fn try_into_outgoing(request: Request) -> Result<(OutgoingRequest, T), Error> { let wasi_req = OutgoingRequest::new(header_map_to_wasi(request.headers())?); diff --git a/src/http/request/sys/p3.rs b/src/http/request/sys/p3.rs new file mode 100644 index 0000000..3d6aec8 --- /dev/null +++ b/src/http/request/sys/p3.rs @@ -0,0 +1,27 @@ +use crate::http::{ + body::{Body, BodyHint}, + error::Error, +}; + +pub use http::request::{Builder, Request}; + +// TODO: go back and add json stuff??? + +pub(crate) fn try_into_outgoing( + request: Request, +) -> Result +where + T: http_body::Body + std::any::Any, + T::Data: Into>, + T::Error: Into>, +{ + wasip3::http_compat::http_into_wasi_request(request).map_err(Into::into) +} + +/// Convert an incoming WASI HTTP request into a wstd request. +#[doc(hidden)] +pub fn try_from_incoming(incoming: wasip3::http::types::Request) -> Result, Error> { + let request = wasip3::http_compat::http_from_wasi_request(incoming)?; + let hint = BodyHint::from_headers(request.headers())?; + Ok(request.map(|body| Body::from_incoming(body, hint))) +} diff --git a/src/http/response.rs b/src/http/response/sys/p2.rs similarity index 100% rename from src/http/response.rs rename to src/http/response/sys/p2.rs diff --git a/src/http/response/sys/p3.rs b/src/http/response/sys/p3.rs new file mode 100644 index 0000000..0c47b89 --- /dev/null +++ b/src/http/response/sys/p3.rs @@ -0,0 +1,30 @@ +use crate::http::body::{Body, BodyHint}; +use crate::http::error::{Error, ErrorCode}; + +pub use http::response::{Builder, Response}; + +pub(crate) fn try_from_incoming( + incoming: wasip3::http::types::Response, +) -> Result, Error> { + let http_response = wasip3::http_compat::http_from_wasi_response(incoming)?; + let hint = BodyHint::from_headers(http_response.headers())?; + Ok(http_response.map(|b| Body::from_incoming(b, hint))) +} + +/// Convert a wstd response into a WASI 0.3 response. +#[doc(hidden)] +pub fn try_into_outgoing(response: Response) -> Result +where + B: Into, +{ + wasip3::http_compat::http_into_wasi_response(response.map(Into::into)).map_err(Into::into) +} + +/// Convert an application error into a WASI HTTP error code. +#[doc(hidden)] +pub fn error_code(error: Error) -> ErrorCode { + error + .downcast_ref::() + .cloned() + .unwrap_or_else(|| ErrorCode::InternalError(Some(format!("{error:?}")))) +} diff --git a/src/http/scheme.rs b/src/http/scheme.rs index 8a3298e..fe05c71 100644 --- a/src/http/scheme.rs +++ b/src/http/scheme.rs @@ -1,7 +1,6 @@ -use wasip2::http::types::Scheme as WasiScheme; - -pub use http::uri::{InvalidUri, Scheme}; +use http::uri::{InvalidUri, Scheme}; use std::str::FromStr; +use wasip2::http::types::Scheme as WasiScheme; pub(crate) fn to_wasi_scheme(value: &Scheme) -> WasiScheme { match value.as_str() { diff --git a/src/http/server.rs b/src/http/server.rs index c9a396c..6bea3e7 100644 --- a/src/http/server.rs +++ b/src/http/server.rs @@ -14,11 +14,11 @@ //! //! [typed main]: https://sunfishcode.github.io/typed-main-wasi-presentation/chapter_1.html //! [`Request`]: crate::http::Request -//! [`Responder`]: crate::http::server::Responder //! [`Response`]: crate::http::Response //! [`http_server`]: crate::http_server -use super::{Body, Error, Response, error::ErrorCode, fields::header_map_to_wasi}; +use super::fields::header_map_to_wasi; +use super::{Body, Error, Response, error::ErrorCode}; use wasip2::exports::http::incoming_handler::ResponseOutparam; use wasip2::http::types::OutgoingResponse; diff --git a/src/io/streams/sys/p3.rs b/src/io/streams/sys/p3.rs index f86e6df..141f5ea 100644 --- a/src/io/streams/sys/p3.rs +++ b/src/io/streams/sys/p3.rs @@ -111,7 +111,7 @@ pub struct AsyncInputChunkStream { enum AsyncInputChunkStreamState { Ready(AsyncInputStream), - Reading(Pin>>), + Reading(Pin + Send>>), Done, } @@ -282,8 +282,7 @@ impl AsyncWrite for AsyncOutputStream { /// # Warning /// /// This is a no-op on generic p3 streams. Use interface-specific flush - /// methods when available (e.g. [`crate::io::Stdout::flush`] or - /// [`crate::io::Stderr::flush`]). + /// methods when available (e.g. `Stdout::flush` or `Stderr::flush`). async fn flush(&mut self) -> std::io::Result<()> { Ok(()) } diff --git a/src/lib.rs b/src/lib.rs index f9c4f06..fc34b46 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,9 +55,7 @@ //! These are unique capabilities provided by WASI 0.2, and because this library //! is specific to that are exposed from here. -#[cfg(all(target_os = "wasi", target_env = "p2"))] pub mod future; -#[cfg(all(target_os = "wasi", target_env = "p2"))] #[macro_use] pub mod http; pub mod io; @@ -70,7 +68,6 @@ pub mod task; #[cfg(target_os = "wasi")] pub mod time; -#[cfg(all(target_os = "wasi", target_env = "p2"))] pub use wstd_macro::attr_macro_http_server as http_server; pub use wstd_macro::{attr_macro_main as main, attr_macro_test as test}; @@ -89,7 +86,6 @@ pub mod __internal { } pub mod prelude { - #[cfg(all(target_os = "wasi", target_env = "p2"))] pub use crate::future::FutureExt as _; pub use crate::io::AsyncRead as _; pub use crate::io::AsyncWrite as _; diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index a92d7fa..72ba783 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -1,8 +1,8 @@ //! Async event loop support. //! -//! On WASI 0.2 the way to use this is to call [`block_on()`]. Inside the -//! future, [`Reactor::current`] will give an instance of the [`Reactor`] -//! running the event loop, which can be used to [`AsyncPollable::wait_for`] +//! On WASI 0.2 the way to use this is to call `block_on()`. Inside the +//! future, `Reactor::current` will give an instance of the `Reactor` +//! running the event loop, which can be used to call `AsyncPollable::wait_for` on //! instances of //! [`wasip2::Pollable`](https://docs.rs/wasi/latest/wasi/io/poll/struct.Pollable.html). //! This will automatically wait for the futures to resolve, and call the diff --git a/src/time/mod.rs b/src/time/mod.rs index b715c11..ed22947 100644 --- a/src/time/mod.rs +++ b/src/time/mod.rs @@ -1,6 +1,5 @@ //! Async time interfaces. -#[cfg(target_env = "p2")] pub(crate) mod utils; mod duration; @@ -172,7 +171,7 @@ mod timer { /// Future created by [`Timer::wait`]. #[must_use = "futures do nothing unless polled or .awaited"] - pub struct Wait(Pin>>); + pub struct Wait(Pin + Send>>); impl Future for Wait { type Output = Instant; diff --git a/test-programs/src/lib.rs b/test-programs/src/lib.rs index bb017da..fb4e32e 100644 --- a/test-programs/src/lib.rs +++ b/test-programs/src/lib.rs @@ -10,13 +10,14 @@ use std::time::Duration; // Required until msrv over 1.89, at which point locking is available in std use fs2::FileExt; -const DEFAULT_SERVER_PORT: u16 = 8081; +const DEFAULT_SERVER_PORT: u16 = 0; /// Manages exclusive access to port 8081, and kills the process when dropped pub struct WasmtimeServe { #[expect(dead_code, reason = "exists to live for as long as wasmtime process")] lockfile: File, process: Child, + addr: Option, } impl WasmtimeServe { @@ -51,18 +52,31 @@ impl WasmtimeServe { for env_var in env_vars { process.arg("--env").arg(env_var); } - let process = process.arg(guest).spawn()?; - let w = WasmtimeServe { lockfile, process }; - + let process = process + .arg(guest) + .stderr(std::process::Stdio::piped()) + .spawn()?; + let mut w = WasmtimeServe { + lockfile, + process, + addr: None, + }; + let listening_addr = get_listening_address(w.process.stderr.as_mut().unwrap()) + .expect("failed to get listening address"); + w.addr = Some(listening_addr); // Clumsily wait for the server to accept connections. 'wait: loop { sleep(Duration::from_millis(100)); - if TcpStream::connect(&listening_addr).is_ok() { + if TcpStream::connect(listening_addr).is_ok() { break 'wait; } } Ok(w) } + + pub fn get_listening_address(&self) -> std::net::SocketAddr { + self.addr.unwrap() + } } // Wasmtime serve will run until killed. Kill it in a drop impl so the process // isnt orphaned when the test suite ends (successfully, or unsuccessfully) @@ -78,10 +92,8 @@ impl Drop for WasmtimeServe { /// Guest programs which bind a socket print `Listening on {addr}`, so that a /// test can discover the address even when the guest picked the port. pub fn get_listening_address( - mut wasmtime_stdout: std::process::ChildStdout, + wasmtime_stdout: &mut impl std::io::Read, ) -> Result { - use std::io::Read; - let mut stdout_contents = String::new(); let mut buf = [0; 4096]; loop { @@ -98,13 +110,14 @@ pub fn get_listening_address( // Parse out the line where guest program says where it is listening for line in stdout_contents.lines() { if let Some(rest) = line.strip_prefix("Listening on ") { - // Forget wasmtime_stdout, rather than drop it, so that any - // subsequent stdout from wasmtime doesn't panic on a broken - // pipe. - std::mem::forget(wasmtime_stdout); return rest .parse() .with_context(|| format!("parsing socket addr from line: {line:?}")); + } else if let Some(rest) = line.strip_prefix("Serving HTTP on http://") { + return rest + .trim_end_matches("/") + .parse() + .with_context(|| format!("parsing http addr from line {line:?}")); } } } diff --git a/test-programs/tests/axum_hello_world.rs b/test-programs/tests/axum_hello_world.rs index 39c782f..c53ce64 100644 --- a/test-programs/tests/axum_hello_world.rs +++ b/test-programs/tests/axum_hello_world.rs @@ -5,22 +5,35 @@ fn hello_world() -> Result<()> { run(test_programs::axum::HELLO_WORLD) } +#[cfg(wstd_nightly)] +#[test_log::test] +fn hello_world_p3() -> Result<()> { + run(test_programs::axum::HELLO_WORLD_P3) +} + #[test_log::test] fn hello_world_nomacro() -> Result<()> { run(test_programs::axum::HELLO_WORLD_NOMACRO) } +#[cfg(wstd_nightly)] +#[test_log::test] +fn hello_world_nomacro_p3() -> Result<()> { + run(test_programs::axum::HELLO_WORLD_NOMACRO_P3) +} + // The hello_world.rs and hello_world_nomacro.rs are identical in // functionality fn run(guest: &str) -> Result<()> { // Run wasmtime serve. - let _serve = test_programs::WasmtimeServe::new(guest)?; + let serve = test_programs::WasmtimeServe::new(guest)?; + let addr = serve.get_listening_address(); // Test each path in the server: // TEST / handler // Response body is the hard-coded default - let body: String = ureq::get("http://127.0.0.1:8081") + let body: String = ureq::get(format!("http://{}", addr)) .call()? .body_mut() .read_to_string()?; diff --git a/test-programs/tests/axum_weather.rs b/test-programs/tests/axum_weather.rs index 4149d0c..a3162f2 100644 --- a/test-programs/tests/axum_weather.rs +++ b/test-programs/tests/axum_weather.rs @@ -6,16 +6,25 @@ const COUNT: usize = 2; #[test_log::test] fn weather() -> Result<()> { + run(test_programs::axum::WEATHER) +} + +#[cfg(wstd_nightly)] +#[test_log::test] +fn weather_p3() -> Result<()> { + run(test_programs::axum::WEATHER_P3) +} + +fn run(component: &str) -> Result<()> { // Run wasmtime serve. - let _serve = test_programs::WasmtimeServe::new(test_programs::axum::WEATHER)?; + let serve = test_programs::WasmtimeServe::new(component)?; + let addr = serve.get_listening_address(); // TEST /weather weather handler - let body = ureq::get(format!( - "http://127.0.0.1:8081/weather?city={CITY}&count={COUNT}" - )) - .call()? - .body_mut() - .read_json::()?; + let body = ureq::get(format!("http://{addr}/weather?city={CITY}&count={COUNT}")) + .call()? + .body_mut() + .read_json::()?; let array = body.as_array().expect("json body is an array"); assert_eq!(array.len(), COUNT); let item_0 = &array[0]; diff --git a/test-programs/tests/http_server.rs b/test-programs/tests/http_server.rs index a4c9283..5c14625 100644 --- a/test-programs/tests/http_server.rs +++ b/test-programs/tests/http_server.rs @@ -3,10 +3,20 @@ use std::time::{Duration, Instant}; #[test_log::test] fn http_server() -> Result<()> { + run(test_programs::HTTP_SERVER) +} + +#[cfg(wstd_nightly)] +#[test_log::test] +fn http_server_p3() -> Result<()> { + run(test_programs::HTTP_SERVER_P3) +} + +fn run(component: &str) -> Result<()> { // Run wasmtime serve. // Enable -Scli because we currently don't have a way to build with the // proxy adapter, so we build with the default adapter. - let _serve = test_programs::WasmtimeServe::new(test_programs::HTTP_SERVER)?; + let _serve = test_programs::WasmtimeServe::new(component)?; // Test each path in the server: diff --git a/test-programs/tests/http_server_proxy.rs b/test-programs/tests/http_server_proxy.rs index 705e2e0..d0b5e9e 100644 --- a/test-programs/tests/http_server_proxy.rs +++ b/test-programs/tests/http_server_proxy.rs @@ -2,10 +2,23 @@ use anyhow::Result; #[test_log::test] fn http_server_proxy() -> Result<()> { + run(test_programs::HTTP_SERVER, test_programs::HTTP_SERVER_PROXY) +} + +#[cfg(wstd_nightly)] +#[test_log::test] +fn http_server_proxy_p3() -> Result<()> { + run( + test_programs::HTTP_SERVER_P3, + test_programs::HTTP_SERVER_PROXY_P3, + ) +} + +fn run(server: &str, proxy: &str) -> Result<()> { // Run wasmtime serve for the proxy and the target HTTP server. - let _serve_target = test_programs::WasmtimeServe::new(test_programs::HTTP_SERVER)?; + let _serve_target = test_programs::WasmtimeServe::new(server)?; let _serve_proxy = test_programs::WasmtimeServe::new_with_config( - test_programs::HTTP_SERVER_PROXY, + proxy, 8082, &["TARGET_URL=http://127.0.0.1:8081"], )?; diff --git a/test-programs/tests/tcp_echo_server.rs b/test-programs/tests/tcp_echo_server.rs index 6e88c1b..befcc18 100644 --- a/test-programs/tests/tcp_echo_server.rs +++ b/test-programs/tests/tcp_echo_server.rs @@ -20,7 +20,7 @@ fn run(component: &str, p3: bool) -> Result<()> { .stdout(std::process::Stdio::piped()) .spawn()?; - let addr = get_listening_address(wasmtime_process.stdout.take().expect("stdout is piped"))?; + let addr = get_listening_address(wasmtime_process.stdout.as_mut().expect("stdout is piped"))?; println!("tcp echo server is listening on {addr:?}"); diff --git a/test-programs/tests/udp_echo_server.rs b/test-programs/tests/udp_echo_server.rs index 8c638cd..137597d 100644 --- a/test-programs/tests/udp_echo_server.rs +++ b/test-programs/tests/udp_echo_server.rs @@ -20,7 +20,7 @@ fn run(component: &str, p3: bool) -> Result<()> { .spawn()?; let addr = test_programs::get_listening_address( - wasmtime_process.stdout.take().expect("stdout is piped"), + wasmtime_process.stdout.as_mut().expect("stdout is piped"), )?; println!("udp echo server is listening on {addr:?}"); diff --git a/tests/http_first_byte_timeout.rs b/tests/http_first_byte_timeout.rs index 11d381a..a77b9fb 100644 --- a/tests/http_first_byte_timeout.rs +++ b/tests/http_first_byte_timeout.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use wstd::http::{Body, Client, Request, error::ErrorCode}; #[wstd::test] diff --git a/tests/http_get.rs b/tests/http_get.rs index 1735d22..e7a3a5a 100644 --- a/tests/http_get.rs +++ b/tests/http_get.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use std::error::Error; use wstd::http::{Body, Client, HeaderValue, Request}; diff --git a/tests/http_get_json.rs b/tests/http_get_json.rs index 641c2b7..a4f42b6 100644 --- a/tests/http_get_json.rs +++ b/tests/http_get_json.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use serde::Deserialize; use std::error::Error; use wstd::http::{Body, Client, Request}; diff --git a/tests/http_handle_error_code.rs b/tests/http_handle_error_code.rs index fa72e0c..6affb90 100644 --- a/tests/http_handle_error_code.rs +++ b/tests/http_handle_error_code.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use wstd::http::{Body, Client, Request, error::ErrorCode}; /// Test that `outgoing_handler::handle` errors are properly propagated. diff --git a/tests/http_post.rs b/tests/http_post.rs index 1082a40..5de184d 100644 --- a/tests/http_post.rs +++ b/tests/http_post.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use std::error::Error; use wstd::http::{Client, HeaderValue, Request}; diff --git a/tests/http_post_json.rs b/tests/http_post_json.rs index f67d050..f9fcf07 100644 --- a/tests/http_post_json.rs +++ b/tests/http_post_json.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use serde::{Deserialize, Serialize}; use std::error::Error; use wstd::http::{Body, Client, HeaderValue, Request}; diff --git a/tests/http_timeout.rs b/tests/http_timeout.rs index 6a156d9..96d40de 100644 --- a/tests/http_timeout.rs +++ b/tests/http_timeout.rs @@ -1,5 +1,3 @@ -#![cfg(all(target_os = "wasi", target_env = "p2"))] - use wstd::future::FutureExt; use wstd::http::{Body, Client, Request}; use wstd::time::Duration;