Implement an HTTP server framework.

Add a wstd API for creating HTTP server applications, wrapping the
WASI proxy world trait and macro.

Compiling the example with `RUSTFLAGS="-Clink-arg=--wasi-adapter=proxy"`
produces a program that runs in `wasmtime serve`.

Author: sunfishcode

Cargo.toml

@@ -15,6 +15,7 @@ categories.workspace = true
 [dependencies]
 futures-core.workspace = true
 http.workspace = true
+itoa.workspace = true
 pin-project-lite.workspace = true
 slab.workspace = true
 wasi.workspace = true
@@ -52,6 +53,7 @@ futures-core = "0.3.19"
 futures-lite = "1.12.0"
 heck = "0.5"
 http = "1.1"
+itoa = "1"
 pin-project-lite = "0.2.8"
 quote = "1.0"
 serde_json = "1"

examples/http_server.rs

@@ -0,0 +1,69 @@
+use wstd::http::body::{BodyForthcoming, IncomingBody, OutgoingBody};
+use wstd::http::server::{Finished, Responder};
+use wstd::http::{IntoBody, Request, Response};
+use wstd::io::{copy, AsyncWrite};
+
+#[wstd::http_server]
+async fn main(request: Request<IncomingBody>, responder: Responder) -> Finished {
+    match request.uri().path_and_query().unwrap().as_str() {
+        "/wait" => http_wait(request, responder).await,
+        "/echo" => http_echo(request, responder).await,
+        "/fail" => http_fail(request, responder).await,
+        "/bigfail" => http_bigfail(request, responder).await,
+        "/" | _ => http_home(request, responder).await,
+    }
+}
+
+async fn http_home(_request: Request<IncomingBody>, responder: Responder) -> Finished {
+    // To send a single string as the response body, use `Responder::respond`.
+    responder
+        .respond(Response::new("Hello, wasi:http/proxy world!\n".into_body()))
+        .await
+}
+
+async fn http_wait(_request: Request<IncomingBody>, responder: Responder) -> Finished {
+    // Get the time now
+    let now = wasi::clocks::monotonic_clock::now();
+
+    // Sleep for 1 second
+    let nanos = 1_000_000_000;
+    let pollable = wasi::clocks::monotonic_clock::subscribe_duration(nanos);
+    pollable.block();
+
+    // Compute how long we slept for.
+    let elapsed = wasi::clocks::monotonic_clock::now() - now;
+    let elapsed = elapsed / 1_000_000; // change to millis
+
+    // To stream data to the response body, use `Responder::start_response`.
+    let mut body = responder.start_response(Response::new(BodyForthcoming));
+    let result = body
+        .write_all(format!("slept for {elapsed} millis\n").as_bytes())
+        .await;
+    Finished::finish(body, result, None)
+}
+
+async fn http_echo(mut request: Request<IncomingBody>, responder: Responder) -> Finished {
+    // Stream data from the request body to the response body.
+    let mut body = responder.start_response(Response::new(BodyForthcoming));
+    let result = copy(request.body_mut(), &mut body).await;
+    Finished::finish(body, result, None)
+}
+
+async fn http_fail(_request: Request<IncomingBody>, responder: Responder) -> Finished {
+    let body = responder.start_response(Response::new(BodyForthcoming));
+    Finished::fail(body)
+}
+
+async fn http_bigfail(_request: Request<IncomingBody>, responder: Responder) -> Finished {
+    async fn write_body(body: &mut OutgoingBody) -> wstd::io::Result<()> {
+        for _ in 0..0x10 {
+            body.write_all("big big big big\n".as_bytes()).await?;
+        }
+        body.flush().await?;
+        Ok(())
+    }
+
+    let mut body = responder.start_response(Response::new(BodyForthcoming));
+    let _ = write_body(&mut body).await;
+    Finished::fail(body)
+}

macro/src/lib.rs

@@ -83,3 +83,75 @@ pub fn attr_macro_test(_attr: TokenStream, item: TokenStream) -> TokenStream {
     }
     .into()
 }
+
+/// Enables a HTTP-server main function, for creating [HTTP servers].
+///
+/// [HTTP servers]: https://docs.rs/wstd/latest/wstd/http/server/index.html
+///
+/// # Examples
+///
+/// ```ignore
+/// #[wstd::http_server]
+/// async fn main(request: Request<IncomingBody>, responder: Responder) -> Finished {
+///     responder
+///         .respond(Response::new("Hello!\n".into_body()))
+///         .await
+/// }
+/// ```
+#[proc_macro_attribute]
+pub fn attr_macro_http_server(_attr: TokenStream, item: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(item as ItemFn);
+
+    if input.sig.asyncness.is_none() {
+        return quote_spanned! { input.sig.fn_token.span()=>
+            compile_error!("fn must be `async fn`");
+        }
+        .into();
+    }
+
+    let output = &input.sig.output;
+    let inputs = &input.sig.inputs;
+    let name = &input.sig.ident;
+    let body = &input.block;
+    let attrs = &input.attrs;
+    let vis = &input.vis;
+
+    if name != "main" {
+        return quote_spanned! { input.sig.ident.span()=>
+            compile_error!("only `async fn main` can be used for #[wstd::http_server]");
+        }
+        .into();
+    }
+
+    quote! {
+        struct TheServer;
+
+        impl ::wstd::wasi::exports::http::incoming_handler::Guest for TheServer {
+            fn handle(
+                request: ::wstd::wasi::http::types::IncomingRequest,
+                response_out: ::wstd::wasi::http::types::ResponseOutparam
+            ) {
+                #(#attrs)*
+                #vis async fn __run(#inputs) #output {
+                    #body
+                }
+
+                let responder = ::wstd::http::server::Responder::new(response_out);
+                let _finished: ::wstd::http::server::Finished =
+                    match ::wstd::http::try_from_incoming_request(request)
+                {
+                    Ok(request) => ::wstd::runtime::block_on(async { __run(request, responder).await }),
+                    Err(err) => responder.fail(err),
+                };
+            }
+        }
+
+        ::wstd::wasi::http::proxy::export!(TheServer with_types_in ::wstd::wasi);
+
+        // In case the user needs it, provide a `main` function so that the
+        // code compiles.
+        #[allow(dead_code)]
+        fn main() { unreachable!("HTTP-server components should be run wth `handle` rather than `main`") }
+    }
+    .into()
+}

src/http/body.rs

@@ -1,6 +1,6 @@
 //! HTTP body types
 
-use crate::io::{AsyncInputStream, AsyncRead, Cursor, Empty};
+use crate::io::{AsyncInputStream, AsyncOutputStream, AsyncRead, AsyncWrite, Cursor, Empty};
 use core::fmt;
 use http::header::{CONTENT_LENGTH, TRANSFER_ENCODING};
 use wasi::http::types::IncomingBody as WasiIncomingBody;
@@ -177,3 +177,79 @@ impl From<InvalidContentLength> for Error {
         ErrorVariant::Other(e.to_string()).into()
     }
 }
+
+/// The output stream for the body, implementing [`AsyncWrite`]. Call
+/// [`Responder::start_response`] to obtain one. Once the body is complete,
+/// it must be declared finished, using [`OutgoingBody::finish`].
+#[must_use]
+pub struct OutgoingBody {
+    // IMPORTANT: the order of these fields here matters. `stream` must
+    // be dropped before `body`.
+    stream: AsyncOutputStream,
+    body: wasi::http::types::OutgoingBody,
+    dontdrop: DontDropOutgoingBody,
+}
+
+impl OutgoingBody {
+    pub(crate) fn new(stream: AsyncOutputStream, body: wasi::http::types::OutgoingBody) -> Self {
+        Self {
+            stream,
+            body,
+            dontdrop: DontDropOutgoingBody,
+        }
+    }
+
+    pub(crate) fn consume(self) -> (AsyncOutputStream, wasi::http::types::OutgoingBody) {
+        let Self {
+            stream,
+            body,
+            dontdrop,
+        } = self;
+
+        std::mem::forget(dontdrop);
+
+        (stream, body)
+    }
+
+    /// Return a reference to the underlying `AsyncOutputStream`.
+    ///
+    /// This usually isn't needed, as `OutgoingBody` implements `AsyncWrite`
+    /// too, however it is useful for code that expects to work with
+    /// `AsyncOutputStream` specifically.
+    pub fn stream(&mut self) -> &mut AsyncOutputStream {
+        &mut self.stream
+    }
+}
+
+impl AsyncWrite for OutgoingBody {
+    async fn write(&mut self, buf: &[u8]) -> crate::io::Result<usize> {
+        self.stream.write(buf).await
+    }
+
+    async fn flush(&mut self) -> crate::io::Result<()> {
+        self.stream.flush().await
+    }
+
+    fn as_async_output_stream(&self) -> Option<&AsyncOutputStream> {
+        Some(&self.stream)
+    }
+}
+
+/// A utility to ensure that `OutgoingBody` is either finished or failed, and
+/// not implicitly dropped.
+struct DontDropOutgoingBody;
+
+impl Drop for DontDropOutgoingBody {
+    fn drop(&mut self) {
+        unreachable!("`OutgoingBody::drop` called; `OutgoingBody`s should be consumed with `finish` or `fail`.");
+    }
+}
+
+/// A placeholder for use as the type parameter to [`Response`] to indicate
+/// that the body has not yet started. This is used with
+/// [`Responder::start_response`], which has a `Response<BodyForthcoming>`
+/// argument.
+///
+/// To instead start the response and obtain the output stream for the body,
+/// use [`Responder::respond`].
+pub struct BodyForthcoming;

src/http/client.rs

@@ -20,6 +20,8 @@ impl Client {
 
     /// Send an HTTP request.
     pub async fn send<B: Body>(&self, req: Request<B>) -> Result<Response<IncomingBody>> {
+        // We don't use `body::OutputBody` here because we can report I/O
+        // errors from the `copy` directly.
         let (wasi_req, body) = try_into_outgoing(req)?;
         let wasi_body = wasi_req.body().unwrap();
         let body_stream = wasi_body.write().unwrap();

src/http/fields.rs

@@ -1,6 +1,9 @@
 pub use http::header::{HeaderMap, HeaderName, HeaderValue};
+use http::header::{InvalidHeaderName, InvalidHeaderValue};
 
+use super::error::ErrorVariant;
 use super::{Error, Result};
+use std::fmt;
 use wasi::http::types::Fields;
 
 pub(crate) fn header_map_from_wasi(wasi_fields: Fields) -> Result<HeaderMap> {
@@ -15,12 +18,52 @@ pub(crate) fn header_map_from_wasi(wasi_fields: Fields) -> Result<HeaderMap> {
     Ok(output)
 }
 
-pub(crate) fn header_map_to_wasi(header_map: &HeaderMap) -> Result<Fields> {
+pub(crate) fn header_map_to_wasi(header_map: &HeaderMap) -> Fields {
     let wasi_fields = Fields::new();
     for (key, value) in header_map {
+        // Unwrap because `HeaderMap` has already validated the headers.
+        // TODO: Remove the `to_owned()` calls after bytecodealliance/wit-bindgen#1102.
         wasi_fields
             .append(&key.as_str().to_owned(), &value.as_bytes().to_owned())
-            .map_err(|e| Error::from(e).context("header named {key}"))?;
+            .unwrap_or_else(|err| panic!("header named {key}: {err:?}"));
+    }
+    wasi_fields
+}
+
+#[derive(Debug)]
+pub(crate) enum InvalidHeader {
+    Name(InvalidHeaderName),
+    Value(InvalidHeaderValue),
+}
+
+impl fmt::Display for InvalidHeader {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Name(e) => e.fmt(f),
+            Self::Value(e) => e.fmt(f),
+        }
+    }
+}
+
+impl std::error::Error for InvalidHeader {}
+
+impl From<InvalidHeaderName> for InvalidHeader {
+    fn from(e: InvalidHeaderName) -> Self {
+        Self::Name(e)
+    }
+}
+
+impl From<InvalidHeaderValue> for InvalidHeader {
+    fn from(e: InvalidHeaderValue) -> Self {
+        Self::Value(e)
+    }
+}
+
+impl From<InvalidHeader> for Error {
+    fn from(e: InvalidHeader) -> Self {
+        match e {
+            InvalidHeader::Name(e) => ErrorVariant::HeaderName(e).into(),
+            InvalidHeader::Value(e) => ErrorVariant::HeaderValue(e).into(),
+        }
     }
-    Ok(wasi_fields)
 }

src/http/method.rs

@@ -1,6 +1,6 @@
 use wasi::http::types::Method as WasiMethod;
 
-use super::Result;
+use http::method::InvalidMethod;
 pub use http::Method;
 
 pub(crate) fn to_wasi_method(value: Method) -> WasiMethod {
@@ -18,9 +18,7 @@ pub(crate) fn to_wasi_method(value: Method) -> WasiMethod {
     }
 }
 
-// This will become useful once we support IncomingRequest
-#[allow(dead_code)]
-pub(crate) fn from_wasi_method(value: WasiMethod) -> Result<Method> {
+pub(crate) fn from_wasi_method(value: WasiMethod) -> Result<Method, InvalidMethod> {
     Ok(match value {
         WasiMethod::Get => Method::GET,
         WasiMethod::Head => Method::HEAD,

src/http/mod.rs

@@ -1,16 +1,17 @@
 //! HTTP networking support
 //!
 pub use http::status::StatusCode;
-pub use http::uri::Uri;
+pub use http::uri::{Authority, PathAndQuery, Uri};
 
 #[doc(inline)]
 pub use body::{Body, IntoBody};
 pub use client::Client;
 pub use error::{Error, Result};
 pub use fields::{HeaderMap, HeaderName, HeaderValue};
 pub use method::Method;
-pub use request::Request;
+pub use request::{try_from_incoming_request, Request};
 pub use response::Response;
+pub use scheme::{InvalidUri, Scheme};
 
 pub mod body;
 
@@ -20,3 +21,5 @@ mod fields;
 mod method;
 mod request;
 mod response;
+mod scheme;
+pub mod server;