Merge pull request #22 from yoshuawuyts/pch/body_impls

Refactor Body impls

Author: pchickey

examples/http_get.rs

@@ -1,5 +1,5 @@
 use std::error::Error;
-use wstd::http::{Client, HeaderValue, Method, Request};
+use wstd::http::{Body, Client, HeaderValue, Method, Request};
 use wstd::io::AsyncRead;
 
 #[wstd::main]
@@ -17,8 +17,19 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .ok_or_else(|| "response expected to have Content-Type header")?;
     assert_eq!(content_type, "application/json; charset=utf-8");
 
+    let body = response.body();
+    let body_len = body
+        .len()
+        .ok_or_else(|| "GET postman-echo.com/get is supposed to provide a content-length")?;
+
     let mut body_buf = Vec::new();
-    let _body_len = response.body().read_to_end(&mut body_buf).await?;
+    body.read_to_end(&mut body_buf).await?;
+
+    assert_eq!(
+        body_buf.len(),
+        body_len,
+        "read_to_end length should match content-length"
+    );
 
     let val: serde_json::Value = serde_json::from_slice(&body_buf)?;
     let body_url = val

src/http/body.rs

@@ -1,6 +1,6 @@
 //! HTTP body types
 
-use crate::io::{AsyncRead, Cursor};
+use crate::io::{AsyncRead, Cursor, Empty};
 
 pub use super::response::IncomingBody;
 
@@ -41,12 +41,24 @@ impl IntoBody for String {
     }
 }
 
-impl<T> Body for T
-where
-    T: AsyncRead,
-{
-    fn len(&self) -> Option<usize> {
-        None
+impl IntoBody for &str {
+    type IntoBody = BoundedBody<Vec<u8>>;
+    fn into_body(self) -> Self::IntoBody {
+        BoundedBody(Cursor::new(self.to_owned().into_bytes()))
+    }
+}
+
+impl IntoBody for Vec<u8> {
+    type IntoBody = BoundedBody<Vec<u8>>;
+    fn into_body(self) -> Self::IntoBody {
+        BoundedBody(Cursor::new(self))
+    }
+}
+
+impl IntoBody for &[u8] {
+    type IntoBody = BoundedBody<Vec<u8>>;
+    fn into_body(self) -> Self::IntoBody {
+        BoundedBody(Cursor::new(self.to_owned()))
     }
 }
 
@@ -59,3 +71,14 @@ impl<T: AsRef<[u8]>> AsyncRead for BoundedBody<T> {
         self.0.read(buf).await
     }
 }
+impl<T: AsRef<[u8]>> Body for BoundedBody<T> {
+    fn len(&self) -> Option<usize> {
+        Some(self.0.get_ref().as_ref().len())
+    }
+}
+
+impl Body for Empty {
+    fn len(&self) -> Option<usize> {
+        Some(0)
+    }
+}

src/http/response.rs

@@ -1,7 +1,7 @@
 use wasi::http::types::{IncomingBody as WasiIncomingBody, IncomingResponse};
 use wasi::io::streams::{InputStream, StreamError};
 
-use super::{fields::header_map_from_wasi, Body, HeaderMap, StatusCode};
+use super::{fields::header_map_from_wasi, Body, Error, HeaderMap, Result, StatusCode};
 use crate::io::AsyncRead;
 use crate::runtime::Reactor;
 
@@ -16,39 +16,36 @@ pub struct Response<B: Body> {
     body: B,
 }
 
-// #[derive(Debug)]
-// enum BodyKind {
-//     Fixed(u64),
-//     Chunked,
-// }
-
-// impl BodyKind {
-//     fn from_headers(headers: &Fields) -> BodyKind {
-//         dbg!(&headers);
-//         if let Some(values) = headers.0.get("content-length") {
-//             let value = values
-//                 .get(0)
-//                 .expect("no value found for content-length; violates HTTP/1.1");
-//             let content_length = String::from_utf8(value.to_owned())
-//                 .unwrap()
-//                 .parse::<u64>()
-//                 .expect("content-length should be a u64; violates HTTP/1.1");
-//             BodyKind::Fixed(content_length)
-//         } else if let Some(values) = headers.0.get("transfer-encoding") {
-//             dbg!(values);
-//             BodyKind::Chunked
-//         } else {
-//             dbg!("Encoding neither has a content-length nor transfer-encoding");
-//             BodyKind::Chunked
-//         }
-//     }
-// }
+#[derive(Debug)]
+enum BodyKind {
+    Fixed(u64),
+    Chunked,
+}
+
+impl BodyKind {
+    fn from_headers(headers: &HeaderMap) -> Result<BodyKind> {
+        if let Some(value) = headers.get("content-length") {
+            let content_length = std::str::from_utf8(value.as_ref())
+                .unwrap()
+                .parse::<u64>()
+                .map_err(|_| {
+                    Error::other("incoming content-length should be a u64; violates HTTP/1.1")
+                })?;
+            Ok(BodyKind::Fixed(content_length))
+        } else if headers.contains_key("transfer-encoding") {
+            Ok(BodyKind::Chunked)
+        } else {
+            Ok(BodyKind::Chunked)
+        }
+    }
+}
 
 impl Response<IncomingBody> {
-    pub(crate) fn try_from_incoming_response(incoming: IncomingResponse) -> super::Result<Self> {
+    pub(crate) fn try_from_incoming_response(incoming: IncomingResponse) -> Result<Self> {
         let headers: HeaderMap = header_map_from_wasi(incoming.headers())?;
         let status = incoming.status().into();
 
+        let kind = BodyKind::from_headers(&headers)?;
         // `body_stream` is a child of `incoming_body` which means we cannot
         // drop the parent before we drop the child
         let incoming_body = incoming
@@ -59,6 +56,7 @@ impl Response<IncomingBody> {
             .expect("cannot call `stream` twice on an incoming body");
 
         let body = IncomingBody {
+            kind,
             buf_offset: 0,
             buf: None,
             body_stream,
@@ -97,6 +95,7 @@ impl<B: Body> Response<B> {
 /// An incoming HTTP body
 #[derive(Debug)]
 pub struct IncomingBody {
+    kind: BodyKind,
     buf: Option<Vec<u8>>,
     // How many bytes have we already read from the buf?
     buf_offset: usize,
@@ -147,3 +146,18 @@ impl AsyncRead for IncomingBody {
         Ok(len)
     }
 }
+
+impl Body for IncomingBody {
+    fn len(&self) -> Option<usize> {
+        match self.kind {
+            BodyKind::Fixed(l) => {
+                if l > (usize::MAX as u64) {
+                    None
+                } else {
+                    Some(l as usize)
+                }
+            }
+            BodyKind::Chunked => None,
+        }
+    }
+}

test-programs/src/bin/http_post.rs

@@ -0,0 +1,54 @@
+use std::error::Error;
+use wstd::http::{Client, HeaderValue, Method, Request};
+use wstd::io::AsyncRead;
+
+#[wstd::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let mut request = Request::new(Method::POST, "https://postman-echo.com/post".parse()?);
+    request.headers_mut().insert(
+        "content-type",
+        HeaderValue::from_str("application/json; charset=utf-8")?,
+    );
+
+    let mut response = Client::new()
+        .send(request.set_body("{\"test\": \"data\"}"))
+        .await?;
+
+    let content_type = response
+        .headers()
+        .get("Content-Type")
+        .ok_or_else(|| "response expected to have Content-Type header")?;
+    assert_eq!(content_type, "application/json; charset=utf-8");
+
+    let mut body_buf = Vec::new();
+    response.body().read_to_end(&mut body_buf).await?;
+
+    let val: serde_json::Value = serde_json::from_slice(&body_buf)?;
+    let body_url = val
+        .get("url")
+        .ok_or_else(|| "body json has url")?
+        .as_str()
+        .ok_or_else(|| "body json url is str")?;
+    assert!(
+        body_url.contains("postman-echo.com/post"),
+        "expected body url to contain the authority and path, got: {body_url}"
+    );
+
+    let posted_json = val
+        .get("json")
+        .ok_or_else(|| "body json has 'json' key")?
+        .as_object()
+        .ok_or_else(|| format!("body json 'json' is object. got {val:?}"))?;
+
+    assert_eq!(posted_json.len(), 1);
+    assert_eq!(
+        posted_json
+            .get("test")
+            .ok_or_else(|| "returned json has 'test' key")?
+            .as_str()
+            .ok_or_else(|| "returned json 'test' key should be str value")?,
+        "data"
+    );
+
+    Ok(())
+}

tests/test-programs.rs

@@ -127,6 +127,13 @@ fn http_get() -> Result<()> {
     run_in_wasmtime(&wasm, None)
 }
 
+#[test]
+fn http_post() -> Result<()> {
+    println!("testing {}", test_programs_artifacts::HTTP_POST);
+    let wasm = std::fs::read(test_programs_artifacts::HTTP_POST).context("read wasm")?;
+    run_in_wasmtime(&wasm, None)
+}
+
 #[test]
 fn http_first_byte_timeout() -> Result<()> {
     println!(