switch up constructors

pchickey · pchickey · commit 658c4eeadec5 · 2025-10-14T15:57:58.000-07:00
diff --git a/examples/complex_http_client.rs b/examples/complex_http_client.rs
@@ -98,7 +98,7 @@ async fn main() -> Result<()> {
             Some(Ok(t))
         }
     });
-    let request = request.body(body)?;
+    let request = request.body(Body::from_http_body(body))?;
 
     // Send the request.
     eprintln!("> {} / {:?}", request.method(), request.version());
diff --git a/examples/http_server.rs b/examples/http_server.rs
@@ -49,18 +49,16 @@ async fn http_wait_body(_request: Request<Body>) -> Result<Response<Body>> {
     // Get the time now
     let now = Instant::now();
 
-    let body = StreamBody::new(once_future(async move {
+    let body = async move {
         // Sleep for one second.
         wstd::task::sleep(Duration::from_secs(1)).await;
 
         // Compute how long we slept for.
         let elapsed = Instant::now().duration_since(now).as_millis();
-        anyhow::Ok(Frame::data(Bytes::from(format!(
-            "slept for {elapsed} millis\n"
-        ))))
-    }));
+        Ok(Bytes::from(format!("slept for {elapsed} millis\n")))
+    };
 
-    Ok(Response::new(body.into()))
+    Ok(Response::new(Body::from_try_stream(once_future(body))))
 }
 
 async fn http_echo(request: Request<Body>) -> Result<Response<Body>> {
@@ -85,7 +83,7 @@ async fn http_echo_trailers(request: Request<Body>) -> Result<Response<Body>> {
     let body = StreamBody::new(once_future(async move {
         anyhow::Ok(Frame::<Bytes>::trailers(trailers))
     }));
-    Ok(Response::new(body.into()))
+    Ok(Response::new(Body::from_http_body(body)))
 }
 
 async fn http_response_status(request: Request<Body>) -> Result<Response<Body>> {
@@ -112,7 +110,7 @@ async fn http_body_fail(_request: Request<Body>) -> Result<Response<Body>> {
         Err::<Frame<Bytes>, _>(anyhow::anyhow!("error creating body"))
     }));
 
-    Ok(Response::new(body.into()))
+    Ok(Response::new(Body::from_http_body(body)))
 }
 
 async fn http_not_found(_request: Request<Body>) -> Result<Response<Body>> {
diff --git a/src/http/body.rs b/src/http/body.rs
@@ -24,6 +24,20 @@ pub mod util {
     pub use http_body_util::*;
 }
 
+/// A HTTP Body.
+///
+/// Construct this HTTP body using:
+/// * `Body::empty` for the empty body, or `impl From<()> for Body`
+/// * `From<&[u8]>` (which will make a clone) or `From<Vec<u8>>` or
+///   `From<Bytes>` for a `Body` from bytes.
+/// * `From<&str>` (which will make a clone) or `From<String>` for a `Body`
+///   from strings.
+/// * `Body::from_json` for a `Body` from a `Serialize` (requires feature
+///   `json`)
+/// * `Body::from_stream` or `Body::from_try_stream` for a `Body` from a
+///   `Stream` of `Into<Bytes>`
+///
+/// Consume
 #[derive(Debug)]
 pub struct Body(BodyInner);
 
@@ -41,7 +55,7 @@ enum BodyInner {
 }
 
 impl Body {
-    pub async fn send(self, outgoing_body: WasiOutgoingBody) -> Result<(), Error> {
+    pub(crate) async fn send(self, outgoing_body: WasiOutgoingBody) -> Result<(), Error> {
         match self.0 {
             BodyInner::Incoming(incoming) => incoming.send(outgoing_body).await,
             BodyInner::Boxed(box_body) => {
@@ -107,17 +121,6 @@ impl Body {
         }
     }
 
-    pub fn as_boxed_body(&mut self) -> &mut UnsyncBoxBody<Bytes, Error> {
-        let mut prev = Self::empty();
-        std::mem::swap(self, &mut prev);
-        self.0 = BodyInner::Boxed(prev.into_boxed_body());
-
-        match &mut self.0 {
-            BodyInner::Boxed(ref mut b) => b,
-            _ => unreachable!(),
-        }
-    }
-
     pub async fn contents(&mut self) -> Result<&[u8], Error> {
         match &mut self.0 {
             BodyInner::Complete { ref data, .. } => Ok(data.as_ref()),
@@ -154,34 +157,22 @@ impl Body {
         }
     }
 
+    /// Construct an empty Body
     pub fn empty() -> Self {
         Body(BodyInner::Complete {
             data: Bytes::new(),
             trailers: None,
         })
     }
 
-    pub fn from_string(s: impl Into<String>) -> Self {
-        let s = s.into();
-        Body(BodyInner::Complete {
-            data: Bytes::from_owner(s.into_bytes()),
-            trailers: None,
-        })
-    }
-
     pub async fn str_contents(&mut self) -> Result<&str, Error> {
         let bs = self.contents().await?;
         std::str::from_utf8(bs).context("decoding body contents as string")
     }
 
-    pub fn from_bytes(b: impl Into<Bytes>) -> Self {
-        let b = b.into();
-        Body::from(http_body_util::Full::new(b))
-    }
-
     #[cfg(feature = "json")]
     pub fn from_json<T: serde::Serialize>(data: &T) -> Result<Self, serde_json::Error> {
-        Ok(Self::from_string(serde_json::to_string(data)?))
+        Ok(Self::from(serde_json::to_vec(data)?))
     }
 
     #[cfg(feature = "json")]
@@ -190,28 +181,38 @@ impl Body {
         serde_json::from_str(str).context("decoding body contents as json")
     }
 
-    pub fn from_input_stream(r: crate::io::AsyncInputStream) -> Self {
-        use futures_lite::stream::StreamExt;
-        Body(BodyInner::Boxed(http_body_util::BodyExt::boxed_unsync(
-            http_body_util::StreamBody::new(r.into_stream().map(|res| {
-                res.map(|bytevec| Frame::data(Bytes::from_owner(bytevec)))
-                    .map_err(Into::into)
-            })),
-        )))
-    }
-
     pub(crate) fn from_incoming(body: WasiIncomingBody, size_hint: BodyHint) -> Self {
         Body(BodyInner::Incoming(Incoming { body, size_hint }))
     }
-}
 
-impl<B> From<B> for Body
-where
-    B: HttpBody + Send + 'static,
-    <B as HttpBody>::Data: Into<Bytes>,
-    <B as HttpBody>::Error: Into<Error>,
-{
-    fn from(http_body: B) -> Body {
+    pub fn from_stream<S>(stream: S) -> Self
+    where
+        S: futures_lite::Stream + Send + 'static,
+        <S as futures_lite::Stream>::Item: Into<Bytes>,
+    {
+        use futures_lite::StreamExt;
+        Self::from_http_body(http_body_util::StreamBody::new(
+            stream.map(|bs| Ok::<_, Error>(Frame::data(bs.into()))),
+        ))
+    }
+
+    pub fn from_try_stream<S, D>(stream: S) -> Self
+    where
+        S: futures_lite::Stream<Item = Result<D, Error>> + Send + 'static,
+        D: Into<Bytes>,
+    {
+        use futures_lite::StreamExt;
+        Self::from_http_body(http_body_util::StreamBody::new(
+            stream.map(|bs| Ok::<_, Error>(Frame::data(bs?.into()))),
+        ))
+    }
+
+    pub fn from_http_body<B>(http_body: B) -> Self
+    where
+        B: HttpBody + Send + 'static,
+        <B as HttpBody>::Data: Into<Bytes>,
+        <B as HttpBody>::Error: Into<Error>,
+    {
         use util::BodyExt;
         Body(BodyInner::Boxed(
             http_body
@@ -222,9 +223,52 @@ where
     }
 }
 
-impl From<Incoming> for Body {
-    fn from(incoming: Incoming) -> Body {
-        Body(BodyInner::Incoming(incoming))
+impl From<()> for Body {
+    fn from(_: ()) -> Body {
+        Body::empty()
+    }
+}
+impl From<&[u8]> for Body {
+    fn from(bytes: &[u8]) -> Body {
+        Body::from(bytes.to_owned())
+    }
+}
+impl From<Vec<u8>> for Body {
+    fn from(bytes: Vec<u8>) -> Body {
+        Body::from(Bytes::from(bytes))
+    }
+}
+impl From<Bytes> for Body {
+    fn from(data: Bytes) -> Body {
+        Body(BodyInner::Complete {
+            data,
+            trailers: None,
+        })
+    }
+}
+impl From<&str> for Body {
+    fn from(data: &str) -> Body {
+        Body::from(data.as_bytes())
+    }
+}
+impl From<String> for Body {
+    fn from(data: String) -> Body {
+        Body::from(data.into_bytes())
+    }
+}
+
+impl From<crate::io::AsyncInputStream> for Body {
+    fn from(r: crate::io::AsyncInputStream) -> Body {
+        // TODO: with another BodyInner variant for a boxed AsyncRead for which
+        // as_input_stream is_some, this could allow for use of
+        // crate::io::copy.
+        use futures_lite::stream::StreamExt;
+        Body(BodyInner::Boxed(http_body_util::BodyExt::boxed_unsync(
+            http_body_util::StreamBody::new(r.into_stream().map(|res| {
+                res.map(|bytevec| Frame::data(Bytes::from_owner(bytevec)))
+                    .map_err(Into::into)
+            })),
+        )))
     }
 }
 
diff --git a/src/io/streams.rs b/src/io/streams.rs
@@ -116,6 +116,18 @@ impl AsyncRead for AsyncInputStream {
     }
 }
 
+#[async_trait::async_trait(?Send)]
+impl AsyncRead for &AsyncInputStream {
+    async fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        Self::read(self, buf).await
+    }
+
+    #[inline]
+    fn as_async_input_stream(&self) -> Option<&AsyncInputStream> {
+        Some(self)
+    }
+}
+
 /// Wrapper of `AsyncInputStream` that impls `futures_lite::stream::Stream`
 /// with an item of `Result<Vec<u8>, std::io::Error>`
 pub struct AsyncInputChunkStream {
diff --git a/tests/http_post.rs b/tests/http_post.rs
@@ -1,5 +1,5 @@
 use std::error::Error;
-use wstd::http::{Body, Client, HeaderValue, Request};
+use wstd::http::{Client, HeaderValue, Request};
 
 #[wstd::test]
 async fn main() -> Result<(), Box<dyn Error>> {
@@ -8,7 +8,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
             "content-type",
             HeaderValue::from_str("application/json; charset=utf-8")?,
         )
-        .body(Body::from_string("{\"test\": \"data\"}"))?;
+        .body("{\"test\": \"data\"}")?;
 
     let response = Client::new().send(request).await?;
 
@@ -19,9 +19,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
     assert_eq!(content_type, "application/json; charset=utf-8");
 
     let mut body = response.into_body();
-    let body_buf = body.contents().await?;
+    let val: serde_json::Value = body.json().await?;
 
-    let val: serde_json::Value = serde_json::from_slice(body_buf)?;
     let body_url = val
         .get("url")
         .ok_or("body json has url")?

Original file line number	Diff line number	Diff line change
`@@ -98,7 +98,7 @@ async fn main() -> Result<()> {`
`98`	`98`	`Some(Ok(t))`
`99`	`99`	`}`
`100`	`100`	`});`
`101`		`- let request = request.body(body)?;`
	`101`	`+ let request = request.body(Body::from_http_body(body))?;`
`102`	`102`
`103`	`103`	`// Send the request.`
`104`	`104`	`eprintln!("> {} / {:?}", request.method(), request.version());`