HTTP streaming client, and trailer support

Cargo.toml

@@ -23,7 +23,9 @@ wstd-macro.workspace = true
 
 [dev-dependencies]
 anyhow.workspace = true
+clap.workspace = true
 futures-lite.workspace = true
+humantime.workspace = true
 serde_json.workspace = true
 
 [workspace]
@@ -44,13 +46,16 @@ categories = []
 authors = [
     "Yoshua Wuyts <[email protected]>",
     "Pat Hickey <[email protected]>",
+    "Dan Gohman <[email protected]>",
 ]
 
 [workspace.dependencies]
 anyhow = "1"
 cargo_metadata = "0.18.1"
+clap = { version = "4.5.26", features = ["derive"] }
 futures-core = "0.3.19"
 futures-lite = "1.12.0"
+humantime = "2.1.0"
 heck = "0.5"
 http = "1.1"
 itoa = "1"

examples/complex_http_client.rs

@@ -0,0 +1,139 @@
+use anyhow::{anyhow, Result};
+use clap::{ArgAction, Parser};
+use std::str::FromStr;
+use wstd::http::{
+    body::BodyForthcoming, Client, HeaderMap, HeaderName, HeaderValue, Method, Request, Uri,
+};
+
+/// Complex HTTP client
+///
+/// A somewhat more complex command-line HTTP client, implemented using
+/// `wstd`, using WASI.
+#[derive(Parser, Debug)]
+#[command(version, about)]
+struct Args {
+    /// The URL to request
+    url: Uri,
+
+    /// Forward stdin to the request body
+    #[arg(long)]
+    body: bool,
+
+    /// Add a header to the request
+    #[arg(long = "header", action = ArgAction::Append, value_name = "HEADER")]
+    headers: Vec<String>,
+
+    /// Add a trailer to the request
+    #[arg(long = "trailer", action = ArgAction::Append, value_name = "TRAILER")]
+    trailers: Vec<String>,
+
+    /// Method of the request
+    #[arg(long, default_value = "GET")]
+    method: Method,
+
+    /// Set the connect timeout
+    #[arg(long, value_name = "DURATION")]
+    connect_timeout: Option<humantime::Duration>,
+
+    /// Set the first-byte timeout
+    #[arg(long, value_name = "DURATION")]
+    first_byte_timeout: Option<humantime::Duration>,
+
+    /// Set the between-bytes timeout
+    #[arg(long, value_name = "DURATION")]
+    between_bytes_timeout: Option<humantime::Duration>,
+}
+
+#[wstd::main]
+async fn main() -> Result<()> {
+    let args = Args::parse();
+
+    // Create and configure the `Client`
+
+    let mut client = Client::new();
+
+    if let Some(connect_timeout) = args.connect_timeout {
+        client.set_connect_timeout(*connect_timeout);
+    }
+    if let Some(first_byte_timeout) = args.first_byte_timeout {
+        client.set_first_byte_timeout(*first_byte_timeout);
+    }
+    if let Some(between_bytes_timeout) = args.between_bytes_timeout {
+        client.set_between_bytes_timeout(*between_bytes_timeout);
+    }
+
+    // Create and configure the request.
+
+    let mut request = Request::builder();
+
+    request = request.uri(args.url).method(args.method);
+
+    for header in args.headers {
+        let mut parts = header.splitn(2, ": ");
+        let key = parts.next().unwrap();
+        let value = parts
+            .next()
+            .ok_or_else(|| anyhow!("headers must be formatted like \"key: value\""))?;
+        request = request.header(key, value);
+    }
+    let mut trailers = HeaderMap::new();
+    for trailer in args.trailers {
+        let mut parts = trailer.splitn(2, ": ");
+        let key = parts.next().unwrap();
+        let value = parts
+            .next()
+            .ok_or_else(|| anyhow!("trailers must be formatted like \"key: value\""))?;
+        trailers.insert(HeaderName::from_str(key)?, HeaderValue::from_str(value)?);
+    }
+
+    // Send the request.
+
+    let request = request.body(BodyForthcoming)?;
+
+    eprintln!("> {} / {:?}", request.method(), request.version());
+    for (key, value) in request.headers().iter() {
+        let value = String::from_utf8_lossy(value.as_bytes());
+        eprintln!("> {key}: {value}");
+    }
+
+    let (mut outgoing_body, response) = client.start_request(request).await?;
+
+    if args.body {
+        wstd::io::copy(wstd::io::stdin(), &mut outgoing_body).await?;
+    } else {
+        wstd::io::copy(wstd::io::empty(), &mut outgoing_body).await?;
+    }
+
+    if !trailers.is_empty() {
+        eprintln!("...");
+    }
+    for (key, value) in trailers.iter() {
+        let value = String::from_utf8_lossy(value.as_bytes());
+        eprintln!("> {key}: {value}");
+    }
+
+    Client::finish(outgoing_body, Some(trailers))?;
+
+    let response = response.await?;
+
+    // Print the response.
+
+    eprintln!("< {:?} {}", response.version(), response.status());
+    for (key, value) in response.headers().iter() {
+        let value = String::from_utf8_lossy(value.as_bytes());
+        eprintln!("< {key}: {value}");
+    }
+
+    let mut body = response.into_body();
+    wstd::io::copy(&mut body, wstd::io::stdout()).await?;
+
+    let trailers = body.finish().await?;
+    if let Some(trailers) = trailers {
+        for (key, value) in trailers.iter() {
+            let value = String::from_utf8_lossy(value.as_bytes());
+            eprintln!("< {key}: {value}");
+        }
+    }
+
+    Ok(())
+}

examples/http_client.rs

@@ -0,0 +1,119 @@
+use anyhow::{anyhow, Result};
+use clap::{ArgAction, Parser};
+use wstd::http::{
+    body::{IncomingBody, StreamedBody},
+    request::Builder,
+    Body, Client, Method, Request, Response, Uri,
+};
+
+/// Simple HTTP client
+///
+/// A simple command-line HTTP client, implemented using `wstd`, using WASI.
+#[derive(Parser, Debug)]
+#[command(version, about)]
+struct Args {
+    /// The URL to request
+    url: Uri,
+
+    /// Forward stdin to the request body
+    #[arg(long)]
+    body: bool,
+
+    /// Add a header to the request
+    #[arg(long = "header", action = ArgAction::Append, value_name = "HEADER")]
+    headers: Vec<String>,
+
+    /// Method of the request
+    #[arg(long, default_value = "GET")]
+    method: Method,
+
+    /// Set the connect timeout
+    #[arg(long, value_name = "DURATION")]
+    connect_timeout: Option<humantime::Duration>,
+
+    /// Set the first-byte timeout
+    #[arg(long, value_name = "DURATION")]
+    first_byte_timeout: Option<humantime::Duration>,
+
+    /// Set the between-bytes timeout
+    #[arg(long, value_name = "DURATION")]
+    between_bytes_timeout: Option<humantime::Duration>,
+}
+
+#[wstd::main]
+async fn main() -> Result<()> {
+    let args = Args::parse();
+
+    // Create and configure the `Client`
+
+    let mut client = Client::new();
+
+    if let Some(connect_timeout) = args.connect_timeout {
+        client.set_connect_timeout(*connect_timeout);
+    }
+    if let Some(first_byte_timeout) = args.first_byte_timeout {
+        client.set_first_byte_timeout(*first_byte_timeout);
+    }
+    if let Some(between_bytes_timeout) = args.between_bytes_timeout {
+        client.set_between_bytes_timeout(*between_bytes_timeout);
+    }
+
+    // Create and configure the request.
+
+    let mut request = Request::builder();
+
+    request = request.uri(args.url).method(args.method);
+
+    for header in args.headers {
+        let mut parts = header.splitn(2, ": ");
+        let key = parts.next().unwrap();
+        let value = parts
+            .next()
+            .ok_or_else(|| anyhow!("headers must be formatted like \"key: value\""))?;
+        request = request.header(key, value);
+    }
+
+    // Send the request.
+
+    async fn send_request<B: Body>(
+        client: &Client,
+        request: Builder,
+        body: B,
+    ) -> Result<Response<IncomingBody>> {
+        let request = request.body(body)?;
+
+        eprintln!("> {} / {:?}", request.method(), request.version());
+        for (key, value) in request.headers().iter() {
+            let value = String::from_utf8_lossy(value.as_bytes());
+            eprintln!("> {key}: {value}");
+        }
+
+        Ok(client.send(request).await?)
+    }
+    let response = if args.body {
+        send_request(&client, request, StreamedBody::new(wstd::io::stdin())).await
+    } else {
+        send_request(&client, request, wstd::io::empty()).await
+    }?;
+
+    // Print the response.
+
+    eprintln!("< {:?} {}", response.version(), response.status());
+    for (key, value) in response.headers().iter() {
+        let value = String::from_utf8_lossy(value.as_bytes());
+        eprintln!("< {key}: {value}");
+    }
+
+    let mut body = response.into_body();
+    wstd::io::copy(&mut body, wstd::io::stdout()).await?;
+
+    let trailers = body.finish().await?;
+    if let Some(trailers) = trailers {
+        for (key, value) in trailers.iter() {
+            let value = String::from_utf8_lossy(value.as_bytes());
+            eprintln!("< {key}: {value}");
+        }
+    }
+
+    Ok(())
+}

src/http/body.rs

@@ -107,6 +107,27 @@ impl<T: AsRef<[u8]>> Body for BoundedBody<T> {
     }
 }
 
+/// An HTTP body with an unknown length
+#[derive(Debug)]
+pub struct StreamedBody<S: AsyncRead>(S);
+
+impl<S: AsyncRead> StreamedBody<S> {
+    /// Wrap an `AsyncRead` impl in a type that provides a [`Body`] implementation.
+    pub fn new(s: S) -> Self {
+        Self(s)
+    }
+}
+impl<S: AsyncRead> AsyncRead for StreamedBody<S> {
+    async fn read(&mut self, buf: &mut [u8]) -> crate::io::Result<usize> {
+        self.0.read(buf).await
+    }
+}
+impl<S: AsyncRead> Body for StreamedBody<S> {
+    fn len(&self) -> Option<usize> {
+        None
+    }
+}
+
 impl Body for Empty {
     fn len(&self) -> Option<usize> {
         Some(0)