wip

Author: pchickey

src/http/body.rs

@@ -24,6 +24,21 @@ 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>`
+/// * `From<AsyncInputStream>`
+///
+/// Consume
 #[derive(Debug)]
 pub struct Body(BodyInner);
 
@@ -33,6 +48,8 @@ enum BodyInner {
     Boxed(UnsyncBoxBody<Bytes, Error>),
     // a body created from a wasi-http incoming-body (WasiIncomingBody)
     Incoming(Incoming),
+    // a body created from an AsyncRead which has a AsyncInputStream
+    InputStream(BoxedInputStream),
     // a body in memory
     Complete {
         data: Bytes,
@@ -41,9 +58,10 @@ 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::InputStream(input_stream) => input_stream.send(outgoing_body).await,
             BodyInner::Boxed(box_body) => {
                 let mut out_stream = AsyncOutputStream::new(
                     outgoing_body
@@ -104,17 +122,7 @@ impl Body {
                 .with_trailers(async move { Ok(trailers).transpose() })
                 .boxed_unsync(),
             BodyInner::Boxed(b) => b,
-        }
-    }
-
-    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!(),
+            BodyInner::InputStream(b) => b.boxed_unsync(),
         }
     }
 
@@ -129,6 +137,7 @@ impl Body {
                 std::mem::swap(inner, &mut prev);
                 let boxed_body = match prev {
                     BodyInner::Incoming(i) => i.into_http_body().boxed_unsync(),
+                    BodyInner::InputStream(i) => i.boxed_unsync(),
                     BodyInner::Boxed(b) => b,
                     BodyInner::Complete { .. } => unreachable!(),
                 };
@@ -151,9 +160,11 @@ impl Body {
             BodyInner::Boxed(b) => b.size_hint().exact(),
             BodyInner::Complete { data, .. } => Some(data.len() as u64),
             BodyInner::Incoming(i) => i.size_hint.content_length(),
+            BodyInner::InputStream(_) => None,
         }
     }
 
+    /// Construct an empty Body
     pub fn empty() -> Self {
         Body(BodyInner::Complete {
             data: Bytes::new(),
@@ -177,20 +188,22 @@ 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 }))
     }
 
+    pub fn from_async_read<R>(read: R) -> Self
+    where
+        R: crate::io::AsyncRead + Send + 'static,
+    {
+        use futures_lite::StreamExt;
+        if read.as_async_input_stream().is_some() {
+            Body(BodyInner::InputStream(BoxedInputStream::new(read)))
+        } else {
+            Self::from_try_stream(read.map(|res| res.map_err(Into::into)))
+        }
+    }
+
     pub fn from_stream<S>(stream: S) -> Self
     where
         S: futures_lite::Stream + Send + 'static,
@@ -202,10 +215,10 @@ impl Body {
         ))
     }
 
-    pub fn from_try_stream<S, B>(stream: S) -> Self
+    pub fn from_try_stream<S, D>(stream: S) -> Self
     where
-        S: futures_lite::Stream<Item = Result<B, Error>> + Send + 'static,
-        B: Into<Bytes>,
+        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(
@@ -263,6 +276,21 @@ impl From<String> for Body {
     }
 }
 
+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)
+            })),
+        )))
+    }
+}
+
 #[derive(Debug)]
 struct Incoming {
     body: WasiIncomingBody,
@@ -549,3 +577,48 @@ impl TrailersState {
         }
     }
 }
+
+struct BoxedInputStream(Box<dyn crate::io::AsyncRead + Send>);
+impl fmt::Debug for BoxedInputStream {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "BoxedInputStream")
+    }
+}
+
+impl BoxedInputStream {
+    fn new(r: impl crate::io::AsyncRead + Send + 'static) -> Self {
+        if r.as_async_input_stream().is_none() {
+            panic!()
+        }
+        Self(Box::new(r))
+    }
+    async fn send(self, outgoing_body: WasiOutgoingBody) -> Result<(), Error> {
+        let mut in_stream = self.0.as_async_input_stream().unwrap();
+        let mut out_stream = AsyncOutputStream::new(
+            outgoing_body
+                .write()
+                .expect("outgoing body already written"),
+        );
+        crate::io::copy(&mut in_stream, &mut out_stream)
+            .await
+            .map_err(|e| {
+                Error::from(e).context("copying incoming body stream to outgoing body stream")
+            })?;
+        drop(in_stream);
+        drop(out_stream);
+        WasiOutgoingBody::finish(outgoing_body, None)
+            .map_err(|e| Error::from(e).context("finishing outgoing body"))?;
+        Ok(())
+    }
+}
+
+impl HttpBody for BoxedInputStream {
+    type Data = Bytes;
+    type Error = Error;
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        todo!()
+    }
+}

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 {