http: convert Request to use AsyncOutputStream

* eliminate its internal OutputStream wrapper.
* fix io::copy to bubble up body copy error rather than panic.
* add http error variant for io errors on body.

Author: pchickey

src/http/client.rs

@@ -1,5 +1,5 @@
 use super::{response::IncomingBody, Body, Error, Request, Response, Result};
-use crate::io::{self, AsyncWrite};
+use crate::io::{self, AsyncOutputStream};
 use crate::runtime::Reactor;
 use crate::time::Duration;
 use wasi::http::types::{OutgoingBody, RequestOptions as WasiRequestOptions};
@@ -27,9 +27,7 @@ impl Client {
         let res = wasi::http::outgoing_handler::handle(wasi_req, self.wasi_options()?).unwrap();
 
         // 2. Start sending the request body
-        io::copy(body, OutputStream::new(body_stream))
-            .await
-            .expect("io::copy broke oh no");
+        io::copy(body, AsyncOutputStream::new(body_stream)).await?;
 
         // 3. Finish sending the request body
         let trailers = None;
@@ -74,33 +72,6 @@ impl Client {
     }
 }
 
-struct OutputStream {
-    stream: wasi::http::types::OutputStream,
-}
-
-impl OutputStream {
-    fn new(stream: wasi::http::types::OutputStream) -> Self {
-        Self { stream }
-    }
-}
-
-impl AsyncWrite for OutputStream {
-    async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let max = self.stream.check_write().unwrap() as usize;
-        let max = max.min(buf.len());
-        let buf = &buf[0..max];
-        self.stream.write(buf).unwrap();
-        Reactor::current().wait_for(self.stream.subscribe()).await;
-        Ok(max)
-    }
-
-    async fn flush(&mut self) -> io::Result<()> {
-        self.stream.flush().unwrap();
-        Reactor::current().wait_for(self.stream.subscribe()).await;
-        Ok(())
-    }
-}
-
 #[derive(Default, Debug)]
 struct RequestOptions {
     connect_timeout: Option<Duration>,

src/http/error.rs

@@ -24,6 +24,7 @@ impl fmt::Debug for Error {
             ErrorVariant::HeaderName(e) => write!(f, "header name error: {e:?}"),
             ErrorVariant::HeaderValue(e) => write!(f, "header value error: {e:?}"),
             ErrorVariant::Method(e) => write!(f, "method error: {e:?}"),
+            ErrorVariant::BodyIo(e) => write!(f, "body error: {e:?}"),
             ErrorVariant::Other(e) => write!(f, "{e}"),
         }
     }
@@ -37,6 +38,7 @@ impl fmt::Display for Error {
             ErrorVariant::HeaderName(e) => write!(f, "header name error: {e}"),
             ErrorVariant::HeaderValue(e) => write!(f, "header value error: {e}"),
             ErrorVariant::Method(e) => write!(f, "method error: {e}"),
+            ErrorVariant::BodyIo(e) => write!(f, "body error: {e}"),
             ErrorVariant::Other(e) => write!(f, "{e}"),
         }
     }
@@ -100,12 +102,19 @@ impl From<InvalidMethod> for Error {
     }
 }
 
+impl From<std::io::Error> for Error {
+    fn from(e: std::io::Error) -> Error {
+        ErrorVariant::BodyIo(e).into()
+    }
+}
+
 #[derive(Debug)]
 pub enum ErrorVariant {
     WasiHttp(WasiHttpErrorCode),
     WasiHeader(WasiHttpHeaderError),
     HeaderName(InvalidHeaderName),
     HeaderValue(InvalidHeaderValue),
     Method(InvalidMethod),
+    BodyIo(std::io::Error),
     Other(String),
 }