feat(io): implement Read and Write for streams

Signed-off-by: Roman Volosatovs <[email protected]>

Author: rvolosatovs

examples/cli-command.rs

@@ -1,11 +1,14 @@
+use std::io::Write as _;
+
 wasi::cli::command::export!(Example);
 
 struct Example;
 
 impl wasi::exports::cli::run::Guest for Example {
     fn run() -> Result<(), ()> {
-        let stdout = wasi::cli::stdout::get_stdout();
-        stdout.blocking_write_and_flush(b"Hello, WASI!").unwrap();
+        let mut stdout = wasi::cli::stdout::get_stdout();
+        stdout.write_all(b"Hello, WASI!").unwrap();
+        stdout.flush().unwrap();
         Ok(())
     }
 }

examples/hello-world.rs

@@ -1,4 +1,7 @@
+use std::io::Write as _;
+
 fn main() {
-    let stdout = wasi::cli::stdout::get_stdout();
-    stdout.blocking_write_and_flush(b"Hello, world!\n").unwrap();
+    let mut stdout = wasi::cli::stdout::get_stdout();
+    stdout.write_all(b"Hello, world!\n").unwrap();
+    stdout.flush().unwrap();
 }

examples/http-proxy.rs

@@ -1,3 +1,5 @@
+use std::io::Write as _;
+
 use wasi::http::types::{
     Fields, IncomingRequest, OutgoingBody, OutgoingResponse, ResponseOutparam,
 };
@@ -13,8 +15,9 @@ impl wasi::exports::http::incoming_handler::Guest for Example {
 
         ResponseOutparam::set(response_out, Ok(resp));
 
-        let out = body.write().unwrap();
-        out.blocking_write_and_flush(b"Hello, WASI!").unwrap();
+        let mut out = body.write().unwrap();
+        out.write_all(b"Hello, WASI!").unwrap();
+        out.flush().unwrap();
         drop(out);
 
         OutgoingBody::finish(body, None).unwrap();

src/errors.rs renamed to src/ext/mod.rs

@@ -1,8 +1,8 @@
+#[cfg(feature = "std")]
+mod std;
+
 impl core::fmt::Display for crate::io::error::Error {
     fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
         f.write_str(&self.to_debug_string())
     }
 }
-
-#[cfg(feature = "std")]
-impl std::error::Error for crate::io::error::Error {}

src/ext/std.rs

@@ -0,0 +1,69 @@
+use std::error::Error;
+use std::io;
+use std::num::NonZeroU64;
+
+use crate::io::streams::StreamError;
+
+impl Error for crate::io::error::Error {}
+
+impl io::Read for crate::io::streams::InputStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        let n = buf
+            .len()
+            .try_into()
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+        match self.blocking_read(n) {
+            Ok(chunk) => {
+                let n = chunk.len();
+                if n > buf.len() {
+                    return Err(io::Error::new(
+                        io::ErrorKind::Other,
+                        "more bytes read than requested",
+                    ));
+                }
+                buf[..n].copy_from_slice(&chunk);
+                Ok(n)
+            }
+            Err(StreamError::Closed) => Ok(0),
+            Err(StreamError::LastOperationFailed(e)) => {
+                Err(io::Error::new(io::ErrorKind::Other, e.to_debug_string()))
+            }
+        }
+    }
+}
+
+impl io::Write for crate::io::streams::OutputStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let n = loop {
+            match self.check_write().map(NonZeroU64::new) {
+                Ok(Some(n)) => {
+                    break n;
+                }
+                Ok(None) => {
+                    self.subscribe().block();
+                }
+                Err(StreamError::Closed) => return Ok(0),
+                Err(StreamError::LastOperationFailed(e)) => {
+                    return Err(io::Error::new(io::ErrorKind::Other, e.to_debug_string()))
+                }
+            };
+        };
+        let n = n
+            .get()
+            .try_into()
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+        let n = buf.len().min(n);
+        crate::io::streams::OutputStream::write(self, &buf[..n]).map_err(|e| match e {
+            StreamError::Closed => io::ErrorKind::UnexpectedEof.into(),
+            StreamError::LastOperationFailed(e) => {
+                io::Error::new(io::ErrorKind::Other, e.to_debug_string())
+            }
+        })?;
+        Ok(n)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.blocking_flush()
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
+    }
+}

src/lib.rs

@@ -179,7 +179,7 @@
 #[cfg(feature = "std")]
 extern crate std;
 
-mod errors;
+mod ext;
 
 // These modules are all auto-generated by `./ci/regenerate.sh`
 mod bindings;