(rust) tcpstream io with vectors

rust/src/main.rs

@@ -32,7 +32,7 @@ async fn main() -> anyhow::Result<()> {
 }
 
 async fn sender_process(running: Sender<bool>) -> anyhow::Result<()> {
-    const BLOCK_SIZE: usize = 32;
+    const BLOCK_SIZE: usize = 128;
     const FILE_PATH: &str = "repositories/test-projects/";
     const FILE_NAME: &str = "small-lorem.md";
 
@@ -62,7 +62,11 @@ async fn sender_process(running: Sender<bool>) -> anyhow::Result<()> {
         }
     };
     let length: usize = file.len();
-    let blocks: usize = (length + BLOCK_SIZE / 2) / BLOCK_SIZE;
+    let blocks: usize = if length % BLOCK_SIZE == 0 {
+        length / BLOCK_SIZE
+    } else {
+        (length / BLOCK_SIZE) + 1
+    };
     let mut index: usize = 0;
 
     connection.writable().await;
@@ -88,38 +92,47 @@ async fn sender_process(running: Sender<bool>) -> anyhow::Result<()> {
 
         sleep(Duration::from_millis(20)).await;
     }
+    println!("send loop done");
 
     while !running.is_closed() {
         sleep(Duration::from_millis(20)).await;
     }
 
+    println!("send loop exiting -> dropping connection");
+
     Ok(())
 }
 
 async fn receiver_process() -> anyhow::Result<()> {
     let mut connection = TcpStream::connect("127.0.0.1:9090").await?;
-    let mut stream: Vec<u8> = Vec::with_capacity(8192);
+    let mut buffer: Vec<u8> = Vec::with_capacity(8192);
     let mut file: Vec<u8> = Vec::new();
     let mut index: usize = 0;
     let mut blocks: usize = 0;
     let mut name: String = "".to_owned();
-    let mut buffer = [0u8; 8192];
+    let mut counter: usize = 3;
-
-    let mut counter: usize = 20;
 
     'receive: loop {
-        let length: usize = stream.len();
+        sleep(Duration::from_millis(20)).await;
+        if counter > 0 || buffer.is_empty() {
+            connection.readable().await;
+            let num_bytes: usize = match connection.read_buf(&mut buffer).await {
+                Ok(0) => {
+                    println!("connection dropped .........");
+                    break 'receive;
+                }
+                Ok(value) => value,
+                Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => continue 'receive,
+                Err(_) => panic!("fuck you!"),
+            };
+        }
 
-        connection.readable().await;
+        if counter > 0 {
-        let num_bytes: usize = match connection.try_read(&mut buffer) {
+            counter -= 1;
-            Ok(0) => continue 'receive,
+            continue 'receive;
-            Ok(value) => value,
+        }
-            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => continue 'receive,
-            Err(_) => panic!("fuck you!"),
-        };
-        stream.extend_from_slice(&buffer[0..num_bytes]);
 
-        let message: FileSection = match Message::from_vec(&mut stream) {
+        let message: FileSection = match Message::from_vec(&mut buffer) {
             Ok(Message::FileSection(value)) => value,
             Ok(unexpected) => panic!(".. {:#?}", unexpected),
             Err(Error::IncompleteFrame) => continue 'receive,
@@ -127,8 +140,6 @@ async fn receiver_process() -> anyhow::Result<()> {
             Err(error) => panic!(".. {:#?}", error),
         };
 
-        // println!("{:#?}", message);
-
         if blocks == 0 {
             blocks = message.sections as usize;
             name = message.name;
@@ -140,9 +151,9 @@ async fn receiver_process() -> anyhow::Result<()> {
         if index >= blocks - 1 {
             break 'receive;
         };
-
-        sleep(Duration::from_millis(20)).await;
     }
+    println!("decoding done");
+    println!("receive loop done");
 
     fs::write("./received.md", file).await;