Unify the read/write arguments (#38)

Default to reading/writing chunks like web. This is also just a safer API, since partial reads are not the normal.

Author: kixelated

web-transport-wasm/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+    println!("cargo:rustc-cfg=web_sys_unstable_apis");
+}

web-transport-wasm/src/recv.rs

@@ -24,40 +24,10 @@ impl RecvStream {
         })
     }
 
-    /// Read some data into the provided buffer.
-    ///
-    /// Returns the (non-zero) number of bytes read, or None if the stream is closed.
-    pub async fn read(&mut self, buf: &mut [u8]) -> Result<Option<usize>, Error> {
-        let chunk = match self.read_chunk(buf.len()).await? {
-            Some(chunk) => chunk,
-            None => return Ok(None),
-        };
-
-        let size = chunk.len();
-        buf[..size].copy_from_slice(&chunk);
-        Ok(Some(size))
-    }
-
-    /// Read some data into the provided buffer.
-    ///
-    /// Returns the (non-zero) number of bytes read, or None if the stream is closed.
-    /// Advances the buffer by the number of bytes read.
-    pub async fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Result<Option<usize>, Error> {
-        let chunk = match self.read_chunk(buf.remaining_mut()).await? {
-            Some(chunk) => chunk,
-            None => return Ok(None),
-        };
-
-        let size = chunk.len();
-        buf.put(chunk);
-
-        Ok(Some(size))
-    }
-
     /// Read the next chunk of data with the provided maximum size.
     ///
     /// This returns a chunk of data instead of copying, which may be more efficient.
-    pub async fn read_chunk(&mut self, max: usize) -> Result<Option<Bytes>, Error> {
+    pub async fn read(&mut self, max: usize) -> Result<Option<Bytes>, Error> {
         if !self.buffer.is_empty() {
             let size = cmp::min(max, self.buffer.len());
             let data = self.buffer.split_to(size).freeze();
@@ -78,6 +48,22 @@ impl RecvStream {
         Ok(Some(data))
     }
 
+    /// Read some data into the provided buffer.
+    ///
+    /// Returns the (non-zero) number of bytes read, or None if the stream is closed.
+    /// Advances the buffer by the number of bytes read.
+    pub async fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Result<Option<usize>, Error> {
+        let chunk = match self.read(buf.remaining_mut()).await? {
+            Some(chunk) => chunk,
+            None => return Ok(None),
+        };
+
+        let size = chunk.len();
+        buf.put(chunk);
+
+        Ok(Some(size))
+    }
+
     /// Abort reading from the stream with the given reason.
     pub fn stop(&mut self, reason: &str) {
         self.reader.abort(reason);

web-transport-wasm/src/send.rs

@@ -16,38 +16,23 @@ impl SendStream {
         Ok(Self { stream, writer })
     }
 
-    /// Write some of the given buffer to the stream.
-    ///
-    /// Returns the non-zero number of bytes written.
-    pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
-        self.writer.write(&Uint8Array::from(buf)).await?;
-        Ok(buf.len())
-    }
-
-    /// Write all of the given buffer to the stream.
-    pub async fn write_all(&mut self, buf: &[u8]) -> Result<(), Error> {
+    /// Write *all* of the given bytes to the stream.
+    pub async fn write(&mut self, buf: &[u8]) -> Result<(), Error> {
         let mut buf = std::io::Cursor::new(buf);
-        self.write_all_buf(&mut buf).await
+        self.write_buf(&mut buf).await
     }
 
-    /// Write some of the given buffer to the stream.
+    /// Write the given buffer to the stream.
     ///
-    /// Returns the non-zero number of bytes written.
     /// Advances the buffer by the number of bytes written.
-    pub async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<usize, Error> {
-        let size = self.write(buf.chunk()).await?;
-        buf.advance(size);
-
-        Ok(size)
-    }
-
-    /// Write all of the given buffer to the stream.
-    ///
-    /// Advances the buffer by the number of bytes written, including any partial writes.
-    pub async fn write_all_buf<B: Buf>(&mut self, buf: &mut B) -> Result<(), Error> {
+    /// May be polled/timed out to perform partial writes.
+    pub async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<(), Error> {
         while buf.has_remaining() {
-            self.write_buf(buf).await?;
+            let chunk = buf.chunk();
+            self.writer.write(&Uint8Array::from(chunk)).await?;
+            buf.advance(chunk.len());
         }
+
         Ok(())
     }
 

web-transport-wasm/src/session.rs

@@ -1,3 +1,4 @@
+use bytes::Bytes;
 use js_sys::{Object, Reflect, Uint8Array};
 use url::Url;
 use wasm_bindgen_futures::JsFuture;
@@ -81,17 +82,17 @@ impl Session {
     }
 
     /// Send a datagram over the network.
-    pub async fn send_datagram(&mut self, payload: &[u8]) -> Result<(), Error> {
+    pub async fn send_datagram(&mut self, payload: Bytes) -> Result<(), Error> {
         let mut writer = Writer::new(&self.inner.datagrams().writable())?;
         writer.write(&Uint8Array::from(payload.as_ref())).await?;
         Ok(())
     }
 
     /// Receive a datagram over the network.
-    pub async fn recv_datagram(&mut self) -> Result<Vec<u8>, Error> {
+    pub async fn recv_datagram(&mut self) -> Result<Bytes, Error> {
         let mut reader = Reader::new(&self.inner.datagrams().readable())?;
         let data: Uint8Array = reader.read().await?.unwrap_or_default();
-        Ok(data.to_vec())
+        Ok(data.to_vec().into())
     }
 
     /// Close the session with the given error code and reason.

web-transport/src/quinn.rs

@@ -96,23 +96,22 @@ impl SendStream {
         Self { inner }
     }
 
-    /// Write some of the buffer to the stream, potentailly blocking on flow control.
-    pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
-        Ok(self.inner.write(buf).await?)
+    /// Write *all* of the buffer to the stream.
+    pub async fn write(&mut self, buf: &[u8]) -> Result<(), Error> {
+        self.inner.write_all(buf).await?;
+        Ok(())
     }
 
-    /// Write some of the given buffer to the stream, potentially blocking on flow control.
-    pub async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<usize, Error> {
-        let size = self.inner.write(buf.chunk()).await?;
-        buf.advance(size);
-        Ok(size)
-    }
-
-    /// Write the entire chunk of bytes to the stream.
+    /// Write the given buffer to the stream, advancing the internal position.
     ///
-    /// More efficient for some implementations, as it avoids a copy
-    pub async fn write_chunk(&mut self, buf: Bytes) -> Result<(), Error> {
-        Ok(self.inner.write_chunk(buf).await?)
+    /// This may be polled to perform partial writes.
+    pub async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<(), Error> {
+        while buf.has_remaining() {
+            let size = self.inner.write(buf.chunk()).await?;
+            buf.advance(size);
+        }
+
+        Ok(())
     }
 
     /// Set the stream's priority.
@@ -141,11 +140,15 @@ impl RecvStream {
         Self { inner }
     }
 
-    /// Read some data into the provided buffer.
+    /// Read the next chunk of data with the provided maximum size.
     ///
-    /// The number of bytes read is returned, or None if the stream is closed.
-    pub async fn read(&mut self, buf: &mut [u8]) -> Result<Option<usize>, Error> {
-        Ok(self.inner.read(buf).await?)
+    /// This returns a chunk of data instead of copying, which may be more efficient.
+    pub async fn read(&mut self, max: usize) -> Result<Option<Bytes>, Error> {
+        Ok(self
+            .inner
+            .read_chunk(max, true)
+            .await?
+            .map(|chunk| chunk.bytes))
     }
 
     /// Read some data into the provided buffer.
@@ -166,17 +169,6 @@ impl RecvStream {
         Ok(Some(size))
     }
 
-    /// Read the next chunk of data with the provided maximum size.
-    ///
-    /// This returns a chunk of data instead of copying, which may be more efficient.
-    pub async fn read_chunk(&mut self, max: usize) -> Result<Option<Bytes>, Error> {
-        Ok(self
-            .inner
-            .read_chunk(max, true)
-            .await?
-            .map(|chunk| chunk.bytes))
-    }
-
     /// Send a `STOP_SENDING` QUIC code.
     pub fn stop(&mut self, code: u32) {
         self.inner.stop(code).ok();

web-transport/src/wasm.rs

@@ -54,21 +54,18 @@ impl From<web_transport_wasm::Session> for Session {
 pub struct SendStream(web_transport_wasm::SendStream);
 
 impl SendStream {
-    pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
+    /// Write all of the given data to the stream.
+    pub async fn write(&mut self, buf: &[u8]) -> Result<(), Error> {
         self.0.write(buf).await
     }
 
     /// Write some of the given buffer to the stream.
-    pub async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<usize, Error> {
+    ///
+    /// Advances the internal position by the number of bytes written.
+    pub async fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Result<(), Error> {
         self.0.write_buf(buf).await
     }
 
-    /// Write the entire chunk of bytes to the stream.
-    /// More efficient for some implementations, as it avoids a copy
-    pub async fn write_chunk(&mut self, buf: Bytes) -> Result<(), Error> {
-        self.0.write_chunk(buf).await
-    }
-
     pub fn set_priority(&mut self, order: i32) {
         self.0.set_priority(order)
     }
@@ -82,20 +79,16 @@ impl SendStream {
 pub struct RecvStream(web_transport_wasm::RecvStream);
 
 impl RecvStream {
-    pub async fn read(&mut self, buf: &mut [u8]) -> Result<Option<usize>, Error> {
-        self.0.read(buf).await
+    /// Attempt to read a chunk of unbuffered data.
+    pub async fn read(&mut self, max: usize) -> Result<Option<Bytes>, Error> {
+        self.0.read(max).await
     }
 
     /// Attempt to read from the stream into the given buffer.
     pub async fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Result<Option<usize>, Error> {
         self.0.read_buf(buf).await
     }
 
-    /// Attempt to read a chunk of unbuffered data.
-    pub async fn read_chunk(&mut self, max: usize) -> Result<Option<Bytes>, Error> {
-        self.0.read_chunk(max).await
-    }
-
     /// Send a `STOP_SENDING` QUIC code.
     pub fn stop(&mut self, code: u32) {
         self.0.stop(&code.to_string())