Back to simple TCP echo server impl and starting from first principles

.gitignore

@@ -8,3 +8,5 @@ Cargo.lock
 
 # These are backup files generated by rustfmt
 **/*.rs.bk
+
+.idea

problem_03/src/client.rs

@@ -1,62 +1,56 @@
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
 use tokio::net::TcpStream;
-use tokio::sync::mpsc::channel;
 use tokio::task;
-use tracing::{debug, info, error};
+use tracing::{error, info};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     tracing_subscriber::fmt::try_init().expect("Tracing was not setup");
 
-    let stream = TcpStream::connect("127.0.0.1:8080").await?;
+    let stream = TcpStream::connect("0.0.0.0:1222").await?;
 
-    let (tx, mut rx) = channel::<String>(10);
+    let (reader, writer) = tokio::io::split(stream);
+    let mut buf_reader = BufReader::new(reader);
+    let mut writer = BufWriter::new(writer);
 
-    let (mut reader, mut writer) = tokio::io::split(stream);
+    let server_handle = task::spawn(async move {
-
+        let mut buf = String::new();
-    let tx_clone = tx.clone();
-
-    task::spawn(async move {
-        let mut reader = BufReader::new(&mut reader);
 
         loop {
             info!("Inside reading lines from server loop");
-            let mut buf = String::new();
+            if let Ok(n) = buf_reader.read_line(&mut buf).await {
-            if let Ok(n) = reader.read_line(&mut buf).await {
                 if n > 0 {
-                    println!("{}", buf.trim_end());
+                    info!("Receivng from server: {}", buf.trim_end());
                 } else {
-                    break;
+                    info!("Server is finished sending, break");
+                    return;
                 }
             } else {
-                break;
+                error!("Cannot receive");
+                return;
             }
+
+            buf.clear();
         }
-        tx_clone.send("exit".to_string()).await.unwrap();
     });
 
-    loop {
+    let std_handle = tokio::spawn(async move {
-        info!("Inside read from std::io loop");
+        let mut stdin_reader = BufReader::new(tokio::io::stdin()).lines();
-        let mut buf = String::new();
+        while let Ok(Some(line)) = stdin_reader.next_line().await {
-        std::io::stdin().read_line(&mut buf)?;
+            info!("Received line from stdin: {}", line);
 
-        let buf = buf.trim_end().to_string();
+            if let Err(_) = writer.write_all(line.as_bytes()).await {
-        info!("New line: {}", buf);
+                error!("Error reading from std");
-        if buf.to_lowercase() == "exit" {
-            break;
-        }
-        debug!(?buf);
-        if let Err(_) = writer.write_all(buf.as_bytes()).await {
-            error!("Could not sent");
                 break;
             }
 
-        if let Some(msg) = rx.recv().await {
+            let _ = writer.write_all(&[b'\n']).await;
-            if msg.to_lowercase() == "exit" {
+            let _ = writer.flush().await;
-                break;
-            }
-        }
         }
+    });
+
+    let _ = server_handle.await;
+    let _ = std_handle.await;
 
     Ok(())
 }

problem_03/src/server.rs

@@ -1,110 +1,73 @@
-use std::collections::HashMap;
+use tokio::{
-use std::sync::{Arc, Mutex};
+    io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter},
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+    net::TcpListener,
-use tokio::net::TcpListener;
+};
-use tokio::sync::broadcast::{self, Sender};
+use tracing::{error, info};
-use tracing::{debug, info, error};
+
+const IP: &str = "0.0.0.0";
+const PORT: u16 = 1222;
+
+type Error = Box<dyn std::error::Error + Send + Sync>;
+type Result<T> = std::result::Result<T, Error>;
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+async fn main() -> Result<()> {
     tracing_subscriber::fmt::try_init().expect("Tracing was not setup");
 
-    let listener = TcpListener::bind("0.0.0.0:1222").await?;
+    let listener = TcpListener::bind(format!("{IP}:{PORT}")).await?;
-    info!("Start listening on 0.0.0.0:1222");
+    info!("Listening on: {}", format!("{IP}:{PORT}"));
+    // Infinite loop to always listen to new connections on this IP/PORT
+    loop {
+        // Get the TCP stream out of the new connection, and the address from which
+        // it is connected to
+        let (mut stream, address) = listener.accept().await?;
+        info!("New address connected: {}", address);
+        // We spawn a new task, so every incoming connection can be put on a thread
+        // and be worked on "in the background"
+        // This allows us to handle multiple connections "at the same time"
+        let _ = stream.write_all("You are connected!\n".as_bytes()).await;
+        tokio::spawn(async move {
+            // From the stream (TcpStream), we can extract the reading, and the writing part
+            // So we can read and write to the connected client on this port
+            let (reader, writer) = stream.split();
 
-    let clients = Arc::new(Mutex::new(HashMap::new()));
+            // So we don't read "directly" on the reader. Therefore we use
-    let (tx, _) = broadcast::channel(10);
+            // BufReader, which performs large, infrequent reads on the underlying
+            // AsyncRead instance (reader)
+            let mut reader = BufReader::new(reader);
+
+            // We do the same for the writing part to the stream
+            // let mut writer = BufWriter::new(writer);
+            let mut writer = BufWriter::new(writer);
+
+            // We need to store what we read from the stream in a local buffer/object
+            let mut line = String::new();
 
             loop {
-        let (mut socket, _addr) = listener.accept().await?;
+                // We read exactly one line per loop. A line ends with \n.
-        let clients = clients.clone();
+                // So if the client doesn't frame their package with \n at the end,
-        let tx = tx.clone();
+                // we won't process until we find one.
-        // let mut stream = stream?;
+                let _ = match reader.read_line(&mut line).await {
+                    Ok(n) if n == 0 => return,
+                    Ok(n) => n,
+                    Err(e) => {
+                        error!("Error reading: {}", e);
+                        return;
+                    }
+                };
 
-        tokio::spawn(async move {
+                info!("New client message received: {}", line.trim_end());
-            let (mut reader, mut writer) = socket.split();
-            let mut buf = String::new();
-            let mut reader = BufReader::new(&mut reader);
 
-            // Request the user's name
+                if let Err(e) = writer.write_all(line.as_bytes()).await {
-            if let Err(e) = writer
+                    error!("Error writing: {}", e);
-                .write_all(b"Welcome to budgetchat! What shall I call you?\n")
-                .await
-            {
-                println!("Failed to send name request: {}", e);
                     return;
                 }
 
-            // Get the user's name
+                let _ = writer.write_all(&[b'\n']).await;
-            match reader.read_line(&mut buf).await {
+                let _ = writer.flush().await;
-                Ok(_) => {
-                    let name = buf.trim().to_string();
-                    info!("Receiving name: {}", name);
-                    if !is_valid_name(&name) {
-                        if let Err(e) = writer
-                            .write_all(b"Invalid name. Connection closed.\n")
-                            .await
-                        {
-                            println!("Failed to send error message: {}", e);
-                        }
-                        return;
-                    }
 
-                    let (client_tx, _client_rx) = broadcast::channel(10);
+                line.clear();
-
-                    {
-                        let mut clients = clients.lock().unwrap();
-                        announce_join(&name, &clients, &tx);
-                        clients.insert(name.clone(), client_tx);
-                    }
-
-                    // Relay messages to other clients
-                    while let Ok(_) = reader.read_line(&mut buf).await {
-                        let message = buf.trim().to_string();
-
-                        if message.is_empty() {
-                            break;
-                        }
-
-                        relay_message(&name, &message, &tx);
-                    }
-
-                    // Client disconnected, remove from clients and announce leave
-                    {
-                        let mut clients = clients.lock().unwrap();
-                        clients.remove(&name);
-                        announce_leave(&name, &clients, &tx);
-                    }
-                }
-                Err(e) => println!("Failed to read name: {}", e),
             }
         });
     }
 }
-
-fn is_valid_name(name: &str) -> bool {
-    !name.is_empty() && name.chars().all(|c| c.is_ascii_alphanumeric())
-}
-
-fn announce_join(name: &str, clients: &HashMap<String, Sender<String>>, tx: &Sender<String>) {
-    let message = format!("* {} has entered the room", name);
-    for client_name in clients.keys() {
-        if client_name != name {
-            let _ = tx.send(message.clone());
-        }
-    }
-}
-
-fn announce_leave(name: &str, clients: &HashMap<String, Sender<String>>, tx: &Sender<String>) {
-    let message = format!("* {} has left the room", name);
-    for client_name in clients.keys() {
-        if client_name != name {
-            let _ = tx.send(message.clone());
-        }
-    }
-}
-
-fn relay_message(name: &str, message: &str, tx: &Sender<String>) {
-    let message = format!("[{}] {}", name, message);
-    let _ = tx.send(message);
-}