feat(http): ensure WasiHttp is Clone + Sync + Send

rvolosatovs · rvolosatovs · commit 9ef58748b446 · 2023-04-24T17:03:11.000+02:00
This should be redundant, but is required by the existing Wasmtime design

Signed-off-by: Roman Volosatovs &lt;rvolosatovs@riseup.net&gt;
diff --git a/crates/wasi-http/src/http_impl.rs b/crates/wasi-http/src/http_impl.rs
@@ -137,7 +137,8 @@ impl WasiHttp {
             // TODO: Configurable default
             Duration::from_millis(600)
         };
-        let mut body = res.into_body();
+        let body = res.into_body();
+        let mut body = body.lock().await;
         let mut buf = BytesMut::new();
         while let Some(next) = timeout(between_bytes_timeout, body.frame()).await? {
             let frame = next?;
diff --git a/crates/wasi-http/src/struct.rs b/crates/wasi-http/src/struct.rs
@@ -9,6 +9,7 @@ use async_trait::async_trait;
 use bytes::{BufMut, Bytes, BytesMut};
 use http_body_util::Full;
 use tokio::net::TcpStream;
+use tokio::sync::Mutex;
 use tokio::time;
 
 #[derive(Clone, Default)]
@@ -24,7 +25,7 @@ impl From<Stream> for Bytes {
 }
 
 #[derive(Clone)]
-pub struct WasiHttp<Response = hyper::body::Incoming> {
+pub struct WasiHttp<Response = Arc<Mutex<hyper::body::Incoming>>> {
     pub outgoing_handler: Arc<Box<dyn OutgoingHandler<Body = Response>>>,
     pub request_id_base: u32,
     pub response_id_base: u32,
@@ -106,7 +107,7 @@ impl From<Bytes> for Stream {
 }
 
 #[async_trait]
-pub trait OutgoingHandler {
+pub trait OutgoingHandler: Sync + Send {
     type Body;
 
     async fn handle(
@@ -123,7 +124,7 @@ pub struct DefaultOutgoingHandler;
 
 #[async_trait]
 impl OutgoingHandler for DefaultOutgoingHandler {
-    type Body = hyper::body::Incoming;
+    type Body = Arc<Mutex<hyper::body::Incoming>>;
 
     async fn handle(
         &self,
@@ -202,6 +203,7 @@ impl OutgoingHandler for DefaultOutgoingHandler {
         time::timeout(first_byte_timeout, sender.send_request(request))
             .await
             .context("request timed out")?
+            .map(|res| res.map(|b| Arc::new(Mutex::new(b))))
             .context("failed to send request")
     }
 }
