v0.4 API 2 with frame acks

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rvideo"
-version = "0.3.4"
+version = "0.4.0"
 edition = "2021"
 authors = ["Serhij S. <[email protected]>"]
 license = "Apache-2.0"

protocol.md

@@ -18,7 +18,8 @@ to communicate with each other.
 
 * Server-to-client: STREAM-INFO
 
-* Server: starts sending frames
+* Server: starts sending frames. To avoid flooding, each frame must be
+  acknowledged by the client before the next one is sent.
 
 ## Structures
 
@@ -102,3 +103,8 @@ For raw formats, the picture length is always the same, however the protocol
 sends the length for each picture data block to allow compressed formats.
 
 The max picture size is `u32::MAX` bytes.
+
+### Acknowledgment
+
+After receiving the frame, the client must send an acknowledgment to the
+server. The acknowledgment is a single byte 0x00.

src/client.rs

@@ -95,6 +95,9 @@ impl Iterator for Client {
         if let Err(e) = self.stream.read_exact(&mut data) {
             return Some(Err(e.into()));
         }
+        if let Err(e) = self.stream.write_all(&[0u8; 1]) {
+            return Some(Err(e.into()));
+        }
         Some(Ok(Frame {
             metadata: metadata.map(Into::into),
             data: data.into(),

src/client_async.rs

@@ -80,6 +80,7 @@ impl ClientAsync {
             usize::try_from(u32::from_le_bytes(len_buf)).map_err(|_| Error::FrameDataTooLarge)?;
         let mut data = vec![0u8; len];
         tokio::time::timeout(self.timeout, self.stream.read_exact(&mut data)).await??;
+        tokio::time::timeout(self.timeout, self.stream.write_all(&[0u8; 1])).await??;
         Ok(Frame {
             metadata: metadata.map(Into::into),
             data: data.into(),

src/lib.rs

@@ -85,7 +85,7 @@ impl Frame {
 }
 
 /// Server API version
-pub const API_VERSION: u8 = 1;
+pub const API_VERSION: u8 = 2;
 
 /// Error type
 #[derive(thiserror::Error, Debug)]

src/server.rs

@@ -232,6 +232,11 @@ impl StreamServerInner {
         let data_len = u32::try_from(frame.data.len()).unwrap();
         socket.write_all(&data_len.to_le_bytes())?;
         socket.write_all(&frame.data)?;
+        let mut buf = [0u8; 1];
+        socket.read_exact(&mut buf)?;
+        if buf[0] != 0 {
+            return Err(Error::NotReady);
+        }
         Ok(())
     }
 }