http: convert IncomingBody to use AsyncInputStream

this eliminates the buffering behavior that IncomingBody previously had. Will
have to define a BufReader for AsyncRead at some point so users can opt back
into that if they want it.

Author: pchickey

src/http/response.rs

@@ -1,12 +1,7 @@
 use wasi::http::types::{IncomingBody as WasiIncomingBody, IncomingResponse};
-use wasi::io::streams::{InputStream, StreamError};
 
 use super::{fields::header_map_from_wasi, Body, Error, HeaderMap, Result, StatusCode};
-use crate::io::AsyncRead;
-use crate::runtime::Reactor;
-
-/// Stream 2kb chunks at a time
-const CHUNK_SIZE: u64 = 2048;
+use crate::io::{AsyncInputStream, AsyncRead};
 
 /// An HTTP response
 #[derive(Debug)]
@@ -57,9 +52,7 @@ impl Response<IncomingBody> {
 
         let body = IncomingBody {
             kind,
-            buf_offset: 0,
-            buf: None,
-            body_stream,
+            body_stream: AsyncInputStream::new(body_stream),
             _incoming_body: incoming_body,
         };
 
@@ -96,54 +89,15 @@ impl<B: Body> Response<B> {
 #[derive(Debug)]
 pub struct IncomingBody {
     kind: BodyKind,
-    buf: Option<Vec<u8>>,
-    // How many bytes have we already read from the buf?
-    buf_offset: usize,
-
     // IMPORTANT: the order of these fields here matters. `body_stream` must
     // be dropped before `_incoming_body`.
-    body_stream: InputStream,
+    body_stream: AsyncInputStream,
     _incoming_body: WasiIncomingBody,
 }
 
 impl AsyncRead for IncomingBody {
     async fn read(&mut self, out_buf: &mut [u8]) -> crate::io::Result<usize> {
-        let buf = match &mut self.buf {
-            Some(ref mut buf) => buf,
-            None => {
-                // Wait for an event to be ready
-                let pollable = self.body_stream.subscribe();
-                Reactor::current().wait_for(pollable).await;
-
-                // Read the bytes from the body stream
-                let buf = match self.body_stream.read(CHUNK_SIZE) {
-                    Ok(buf) => buf,
-                    Err(StreamError::Closed) => return Ok(0),
-                    Err(StreamError::LastOperationFailed(err)) => {
-                        return Err(std::io::Error::other(format!(
-                            "last operation failed: {}",
-                            err.to_debug_string()
-                        )))
-                    }
-                };
-                self.buf.insert(buf)
-            }
-        };
-
-        // copy bytes
-        let len = (buf.len() - self.buf_offset).min(out_buf.len());
-        let max = self.buf_offset + len;
-        let slice = &buf[self.buf_offset..max];
-        out_buf[0..len].copy_from_slice(slice);
-        self.buf_offset += len;
-
-        // reset the local slice if necessary
-        if self.buf_offset == buf.len() {
-            self.buf = None;
-            self.buf_offset = 0;
-        }
-
-        Ok(len)
+        self.body_stream.read(out_buf).await
     }
 }
 

src/io/stdio.rs

@@ -4,6 +4,7 @@ use wasi::cli::terminal_input::TerminalInput;
 use wasi::cli::terminal_output::TerminalOutput;
 
 /// Use the program's stdin as an `AsyncInputStream`.
+#[derive(Debug)]
 pub struct Stdin {
     stream: AsyncInputStream,
     terminput: LazyCell<Option<TerminalInput>>,
@@ -38,6 +39,7 @@ impl Stdin {
 }
 
 /// Use the program's stdout as an `AsyncOutputStream`.
+#[derive(Debug)]
 pub struct Stdout {
     stream: AsyncOutputStream,
     termoutput: LazyCell<Option<TerminalOutput>>,
@@ -72,6 +74,7 @@ impl std::ops::DerefMut for Stdout {
 }
 
 /// Use the program's stdout as an `AsyncOutputStream`.
+#[derive(Debug)]
 pub struct Stderr {
     stream: AsyncOutputStream,
     termoutput: LazyCell<Option<TerminalOutput>>,

src/io/streams.rs

@@ -3,6 +3,7 @@ use std::cell::RefCell;
 use std::io::Result;
 use wasi::io::streams::{InputStream, OutputStream, StreamError};
 
+#[derive(Debug)]
 pub struct AsyncInputStream {
     // Lazily initialized pollable, used for lifetime of stream to check readiness.
     // Field ordering matters: this child must be dropped before stream
@@ -51,6 +52,7 @@ impl AsyncRead for AsyncInputStream {
     }
 }
 
+#[derive(Debug)]
 pub struct AsyncOutputStream {
     // Lazily initialized pollable, used for lifetime of stream to check readiness.
     // Field ordering matters: this child must be dropped before stream