Allow sending OutgoingRequest bodies

Signed-off-by: Ryan Levick <[email protected]>

Author: rylev

sdk/rust/src/http.rs

@@ -5,16 +5,17 @@ use std::collections::HashMap;
 
 #[doc(inline)]
 pub use conversions::IntoResponse;
-
-use self::conversions::TryFromIncomingResponse;
-
-use super::wit::wasi::http::types;
 #[doc(inline)]
 pub use types::{
     Error, Fields, Headers, IncomingRequest, IncomingResponse, Method, OutgoingBody,
     OutgoingRequest, OutgoingResponse, Scheme, StatusCode, Trailers,
 };
 
+use self::conversions::{TryFromIncomingResponse, TryIntoOutgoingRequest};
+use super::wit::wasi::http::types;
+use crate::wit::wasi::io::streams;
+use futures::SinkExt;
+
 /// A unified request object that can represent both incoming and outgoing requests.
 ///
 /// This should be used in favor of `IncomingRequest` and `OutgoingRequest` when there
@@ -438,12 +439,12 @@ impl IncomingRequest {
     /// # Panics
     ///
     /// Panics if the body was already consumed.
-    pub fn into_body_stream(self) -> impl futures::Stream<Item = anyhow::Result<Vec<u8>>> {
+    pub fn into_body_stream(self) -> impl futures::Stream<Item = Result<Vec<u8>, streams::Error>> {
         executor::incoming_body(self.consume().expect("request body was already consumed"))
     }
 
     /// Return a `Vec<u8>` of the body or fails
-    pub async fn into_body(self) -> anyhow::Result<Vec<u8>> {
+    pub async fn into_body(self) -> Result<Vec<u8>, streams::Error> {
         use futures::TryStreamExt;
         let mut stream = self.into_body_stream();
         let mut body = Vec::new();
@@ -460,12 +461,12 @@ impl IncomingResponse {
     /// # Panics
     ///
     /// Panics if the body was already consumed.
-    pub fn into_body_stream(self) -> impl futures::Stream<Item = anyhow::Result<Vec<u8>>> {
+    pub fn into_body_stream(self) -> impl futures::Stream<Item = Result<Vec<u8>, streams::Error>> {
         executor::incoming_body(self.consume().expect("response body was already consumed"))
     }
 
     /// Return a `Vec<u8>` of the body or fails
-    pub async fn into_body(self) -> anyhow::Result<Vec<u8>> {
+    pub async fn into_body(self) -> Result<Vec<u8>, streams::Error> {
         use futures::TryStreamExt;
         let mut stream = self.into_body_stream();
         let mut body = Vec::new();
@@ -487,6 +488,17 @@ impl OutgoingResponse {
     }
 }
 
+impl OutgoingRequest {
+    /// Construct a `Sink` which writes chunks to the body of the specified response.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the body was already taken.
+    pub fn take_body(&self) -> impl futures::Sink<Vec<u8>, Error = Error> {
+        executor::outgoing_body(self.write().expect("request body was already taken"))
+    }
+}
+
 /// The out param for setting an `OutgoingResponse`
 pub struct ResponseOutparam(types::ResponseOutparam);
 
@@ -511,7 +523,6 @@ impl ResponseOutparam {
         response: OutgoingResponse,
         buffer: Vec<u8>,
     ) -> Result<(), Error> {
-        use futures::SinkExt;
         let mut body = response.take_body();
         self.set(response);
         body.send(buffer).await
@@ -526,18 +537,26 @@ impl ResponseOutparam {
 /// Send an outgoing request
 pub async fn send<I, O>(request: I) -> Result<O, SendError>
 where
-    I: TryInto<OutgoingRequest>,
+    I: TryIntoOutgoingRequest,
     I::Error: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
     O: TryFromIncomingResponse,
     O::Error: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
 {
-    let response = executor::outgoing_request_send(
-        request
-            .try_into()
-            .map_err(|e| SendError::RequestConversion(e.into()))?,
-    )
-    .await
-    .map_err(SendError::Http)?;
+    let (request, body_buffer) = I::try_into_outgoing_request(request)
+        .map_err(|e| SendError::RequestConversion(e.into()))?;
+    if let Some(body_buffer) = body_buffer {
+        // It is part of the contract of the trait that implementors of `TryIntoOutgoingRequest`
+        // do not call `OutgoingRequest::write`` if they return a buffered body.
+        let mut body_sink = request.take_body();
+        body_sink
+            .send(body_buffer)
+            .await
+            .map_err(|e| SendError::Http(Error::UnexpectedError(e.to_string())))?;
+    }
+    let response = executor::outgoing_request_send(request)
+        .await
+        .map_err(SendError::Http)?;
+
     TryFromIncomingResponse::try_from_incoming_response(response)
         .await
         .map_err(|e: O::Error| SendError::ResponseConversion(e.into()))
@@ -642,6 +661,14 @@ pub mod responses {
     }
 }
 
+impl std::fmt::Display for streams::Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.to_debug_string())
+    }
+}
+
+impl std::error::Error for streams::Error {}
+
 #[cfg(test)]
 mod tests {
     use super::*;

sdk/rust/src/http/conversions.rs

@@ -2,6 +2,8 @@ use std::collections::HashMap;
 
 use async_trait::async_trait;
 
+use crate::wit::wasi::io::streams;
+
 use super::{Headers, IncomingRequest, IncomingResponse, OutgoingRequest, OutgoingResponse};
 
 use super::{responses, NonUtf8BodyError, Request, Response};
@@ -61,12 +63,12 @@ impl TryFromIncomingRequest for Request {
             .method(request.method())
             .uri(request.uri())
             .headers(request.headers())
-            .body(
-                request
-                    .into_body()
-                    .await
-                    .map_err(IncomingRequestError::BodyConversionError)?,
-            )
+            .body(request.into_body().await.map_err(|e| {
+                IncomingRequestError::BodyConversionError(anyhow::anyhow!(
+                    "{}",
+                    e.to_debug_string()
+                ))
+            })?)
             .build())
     }
 }
@@ -482,37 +484,59 @@ where
     }
 }
 
-impl TryFrom<Request> for OutgoingRequest {
+/// A trait for converting a type into an `OutgoingRequest`
+pub trait TryIntoOutgoingRequest {
+    /// The error if the conversion fails
+    type Error;
+
+    /// Turn the type into an `OutgoingRequest`
+    ///
+    /// If the implementor can be sure that the `OutgoingRequest::write` has not been called they
+    /// can return a buffer as the second element of the returned tuple and `send` will send
+    /// that as the request body.
+    fn try_into_outgoing_request(self) -> Result<(OutgoingRequest, Option<Vec<u8>>), Self::Error>;
+}
+
+impl TryIntoOutgoingRequest for OutgoingRequest {
+    type Error = std::convert::Infallible;
+
+    fn try_into_outgoing_request(self) -> Result<(OutgoingRequest, Option<Vec<u8>>), Self::Error> {
+        Ok((self, None))
+    }
+}
+
+impl TryIntoOutgoingRequest for Request {
     type Error = std::convert::Infallible;
 
-    fn try_from(req: Request) -> Result<Self, Self::Error> {
-        let headers = req
+    fn try_into_outgoing_request(self) -> Result<(OutgoingRequest, Option<Vec<u8>>), Self::Error> {
+        let headers = self
             .headers()
             .map(|(k, v)| (k.to_owned(), v.as_bytes().to_owned()))
             .collect::<Vec<_>>();
-        Ok(OutgoingRequest::new(
-            req.method(),
-            req.path_and_query(),
-            Some(if req.is_https() {
+        let request = OutgoingRequest::new(
+            self.method(),
+            self.path_and_query(),
+            Some(if self.is_https() {
                 &super::Scheme::Https
             } else {
                 &super::Scheme::Http
             }),
-            req.authority(),
+            self.authority(),
             &Headers::new(&headers),
-        ))
+        );
+        Ok((request, Some(self.into_body())))
     }
 }
 
 #[cfg(feature = "http")]
-impl<B> TryFrom<hyperium::Request<B>> for OutgoingRequest
+impl<B> TryIntoOutgoingRequest for hyperium::Request<B>
 where
     B: TryIntoBody,
     B::Error: std::error::Error + Send + Sync + 'static,
 {
     type Error = anyhow::Error;
-    fn try_from(req: hyperium::Request<B>) -> Result<Self, Self::Error> {
-        let method = match req.method() {
+    fn try_into_outgoing_request(self) -> Result<(OutgoingRequest, Option<Vec<u8>>), Self::Error> {
+        let method = match self.method() {
             &hyperium::Method::GET => super::Method::Get,
             &hyperium::Method::POST => super::Method::Post,
             &hyperium::Method::PUT => super::Method::Put,
@@ -522,34 +546,36 @@ where
             &hyperium::Method::OPTIONS => super::Method::Options,
             m => anyhow::bail!("Unsupported method: {m}"),
         };
-        let headers = req
+        let headers = self
             .headers()
             .into_iter()
             .map(|(n, v)| (n.as_str().to_owned(), v.as_bytes().to_owned()))
             .collect::<Vec<_>>();
-        Ok(OutgoingRequest::new(
+        let request = OutgoingRequest::new(
             &method,
-            req.uri().path_and_query().map(|p| p.as_str()),
-            req.uri()
+            self.uri().path_and_query().map(|p| p.as_str()),
+            self.uri()
                 .scheme()
                 .map(|s| match s.as_str() {
                     "http" => super::Scheme::Http,
                     "https" => super::Scheme::Https,
                     s => super::Scheme::Other(s.to_owned()),
                 })
                 .as_ref(),
-            req.uri().authority().map(|a| a.as_str()),
+            self.uri().authority().map(|a| a.as_str()),
             &Headers::new(&headers),
-        ))
+        );
+        let buffer = TryIntoBody::try_into_body(self.into_body())?;
+        Ok((request, Some(buffer)))
     }
 }
 
+/// A trait for converting from an `IncomingRequest`
 #[async_trait]
-/// TODO
 pub trait TryFromIncomingResponse {
-    /// TODO
+    /// The error if conversion fails
     type Error;
-    /// TODO
+    /// Turn the `IncomingResponse` into the type
     async fn try_from_incoming_response(resp: IncomingResponse) -> Result<Self, Self::Error>
     where
         Self: Sized;
@@ -563,6 +589,18 @@ impl TryFromIncomingResponse for IncomingResponse {
     }
 }
 
+#[async_trait]
+impl TryFromIncomingResponse for Response {
+    type Error = streams::Error;
+    async fn try_from_incoming_response(resp: IncomingResponse) -> Result<Self, Self::Error> {
+        Ok(Response::builder()
+            .status(resp.status())
+            .headers(resp.headers())
+            .body(resp.into_body().await?)
+            .build())
+    }
+}
+
 #[cfg(feature = "http")]
 #[async_trait]
 impl<B: TryFromBody> TryFromIncomingResponse for hyperium::Response<B> {

sdk/rust/src/http/executor.rs

@@ -5,7 +5,6 @@ use crate::wit::wasi::http::types::{
 use crate::wit::wasi::io;
 use crate::wit::wasi::io::streams::{InputStream, OutputStream, StreamError};
 
-use anyhow::{anyhow, Result};
 use futures::{future, sink, stream, Sink, Stream};
 
 use std::cell::RefCell;
@@ -165,7 +164,9 @@ pub(crate) fn outgoing_request_send(
 }
 
 #[doc(hidden)]
-pub fn incoming_body(body: IncomingBody) -> impl Stream<Item = Result<Vec<u8>>> {
+pub fn incoming_body(
+    body: IncomingBody,
+) -> impl Stream<Item = Result<Vec<u8>, io::streams::Error>> {
     struct Incoming(Option<(InputStream, IncomingBody)>);
 
     impl Drop for Incoming {
@@ -196,9 +197,7 @@ pub fn incoming_body(body: IncomingBody) -> impl Stream<Item = Result<Vec<u8>>>
                         }
                     }
                     Err(StreamError::Closed) => Poll::Ready(None),
-                    Err(StreamError::LastOperationFailed(error)) => Poll::Ready(Some(Err(
-                        anyhow!("Last operation failed: {}", error.to_debug_string()),
-                    ))),
+                    Err(StreamError::LastOperationFailed(error)) => Poll::Ready(Some(Err(error))),
                 }
             } else {
                 Poll::Ready(None)