Fix streaming request bodies

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

Author: rylev

examples/http-rust-outbound-http/outbound-http-to-same-app/src/lib.rs

@@ -7,7 +7,7 @@ use spin_sdk::{
 /// Send an HTTP request and return the response.
 #[http_component]
 async fn send_outbound(_req: Request) -> Result<impl IntoResponse> {
-    let mut res: http::Response<()> = spin_sdk::http::send(
+    let mut res: http::Response<String> = spin_sdk::http::send(
         http::Request::builder()
             .method("GET")
             .uri("/hello")

sdk/rust/readme.md

@@ -41,7 +41,7 @@ server, modifies the result, then returns it:
 ```rust
 #[http_component]
 async fn hello_world(_req: Request) -> Result<Response> {
-    let mut res: http::Response<()> = spin_sdk::http::send(
+    let mut res: http::Response<String> = spin_sdk::http::send(
         http::Request::builder()
             .method("GET")
             .uri("https://fermyon.com")

sdk/rust/src/http.rs

@@ -535,6 +535,9 @@ impl ResponseOutparam {
 }
 
 /// Send an outgoing request
+///
+/// If `request`` is an `OutgoingRequest` and you are streaming the body to the
+/// outgoing request body sink, you need to ensure it is dropped before awaiting this function.
 pub async fn send<I, O>(request: I) -> Result<O, SendError>
 where
     I: TryIntoOutgoingRequest,
@@ -544,18 +547,22 @@ where
 {
     let (request, body_buffer) = I::try_into_outgoing_request(request)
         .map_err(|e| SendError::RequestConversion(e.into()))?;
-    if let Some(body_buffer) = body_buffer {
+    let response = 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();
+        let response = executor::outgoing_request_send(request);
         body_sink
             .send(body_buffer)
             .await
             .map_err(|e| SendError::Http(Error::UnexpectedError(e.to_string())))?;
+        // The body sink needs to be dropped before we await the response, otherwise we deadlock
+        drop(body_sink);
+        response.await
+    } else {
+        executor::outgoing_request_send(request).await
     }
-    let response = executor::outgoing_request_send(request)
-        .await
-        .map_err(SendError::Http)?;
+    .map_err(SendError::Http)?;
 
     TryFromIncomingResponse::try_from_incoming_response(response)
         .await

sdk/rust/src/http/executor.rs

@@ -143,9 +143,8 @@ pub(crate) fn outgoing_body(body: OutgoingBody) -> impl Sink<Vec<u8>, Error = ty
 pub(crate) fn outgoing_request_send(
     request: OutgoingRequest,
 ) -> impl Future<Output = Result<IncomingResponse, types::Error>> {
+    let response = outgoing_handler::handle(request, None);
     future::poll_fn({
-        let response = outgoing_handler::handle(request, None);
-
         move |context| match &response {
             Ok(response) => {
                 if let Some(response) = response.get() {