Chat Server

We build a chat server that anyone can join with nc or telnet. Every line a user types is sent to everyone else, and a few commands start with /:

Text

$ nc localhost 7878
welcome! you are guest-1. commands: /nick <name>, /list, /quit
/nick ana
you are now ana
guest-2 joined
guest-2 is now ben
hi ben
ben: hi ana!

This is the classic exercise for asynchronous I/O. A traditional server would use one operating system thread per connection; with Tokio each connection is a lightweight task.

Step	Topics
1. the protocol	an enum of commands, parsing with `strip_prefix` and `split_once`, unit tests
2. fan-out	`tokio::sync::broadcast`: every subscriber gets every message, and slow ones are told what they missed
3. the server	`TcpListener`, one task per connection, `select!` over socket input and broadcasts, `Arc<Mutex<...>>` for nicknames
4. testing it	scripted clients over real TCP, waiting for each expected line

Terminal

cargo new chat
cd chat
cargo add tokio --features full

Step 1: Parsing what users type

Parsing is kept separate from networking, so the protocol can be tested without sockets. Anything that doesn't start with / is a message.

Rustcargo test

#[derive(Debug, PartialEq)]
pub enum Command<'a> {
    Say(&'a str),
    Nick(&'a str),
    List,
    Quit,
    Invalid(String),
}

pub fn parse(line: &str) -> Command<'_> {
    let line = line.trim();
    let Some(command) = line.strip_prefix('/') else {
        return Command::Say(line);
    };
    let (name, argument) = command.split_once(' ').map_or((command, ""), |(n, a)| (n, a.trim()));
    match (name, argument) {
        ("nick", "") => Command::Invalid(String::from("usage: /nick <name>")),
        ("nick", nick) if nick.contains(' ') || nick.len() > 16 => {
            Command::Invalid(String::from("names are one word, at most 16 characters"))
        }
        ("nick", nick) => Command::Nick(nick),
        ("list", _) => Command::List,
        ("quit", _) => Command::Quit,
        (other, _) => Command::Invalid(format!("unknown command /{other}")),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn plain_lines_are_messages() {
        assert_eq!(parse("  hello there  "), Command::Say("hello there"));
    }

    #[test]
    fn commands() {
        assert_eq!(parse("/nick ana"), Command::Nick("ana"));
        assert_eq!(parse("/list"), Command::List);
        assert_eq!(parse("/quit now"), Command::Quit);
    }

    #[test]
    fn bad_commands_explain_themselves() {
        assert_eq!(parse("/nick"), Command::Invalid(String::from("usage: /nick <name>")));
        assert_eq!(parse("/nick ana maria"), Command::Invalid(String::from("names are one word, at most 16 characters")));
        assert_eq!(parse("/shout hi"), Command::Invalid(String::from("unknown command /shout")));
    }
}

Output

running 3 tests
test tests::bad_commands_explain_themselves ... ok
test tests::commands ... ok
test tests::plain_lines_are_messages ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Command<'a> borrows from the input line: Say and Nick hold &str slices, so parsing allocates nothing.

Step 2: Fanning out with a broadcast channel

Every message must reach every connected client. tokio::sync::broadcast is made for this: each connection calls subscribe() to get its own receiver, and every value sent is delivered to all current receivers.

The channel has a fixed capacity. A client that reads too slowly doesn't block everyone else; instead, when it falls more than capacity messages behind, its next recv() returns Lagged(n), and the oldest messages are gone for that receiver only.

Rust

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (sender, mut fast) = broadcast::channel::<String>(4);
    let mut slow = sender.subscribe();

    for n in 1..=3 {
        sender.send(format!("message {n}")).unwrap();
        println!("fast got {:?}", fast.recv().await.unwrap());
    }

    // `slow` hasn't read anything. Three more messages overflow its capacity of 4.
    for n in 4..=6 {
        sender.send(format!("message {n}")).unwrap();
        fast.recv().await.unwrap();
    }
    loop {
        match slow.try_recv() {
            Ok(message) => println!("slow got {message:?}"),
            Err(broadcast::error::TryRecvError::Lagged(missed)) => println!("slow missed {missed} messages"),
            Err(_) => break, // empty
        }
    }

    drop(sender);
    println!("after the sender is dropped: {:?}", fast.recv().await);
}

Output

fast got "message 1"
fast got "message 2"
fast got "message 3"
slow missed 2 messages
slow got "message 3"
slow got "message 4"
slow got "message 5"
slow got "message 6"
after the sender is dropped: Err(Closed)

For a chat that's the right trade-off: a stuck client loses some messages and is told so, while everyone else carries on. (mpsc would be wrong here: each message goes to exactly one receiver.)

Step 3 and 4: The server, tested with real clients

Each connection runs this loop in its own task:

Rust

loop {
    tokio::select! {
        line = lines.next_line() => { /* the user typed something: handle the command */ }
        event = events.recv() => { /* someone else said something: write it to this user */ }
    }
}

select! waits on both futures at once and runs the branch of whichever finishes first. A thread-per-connection server would need two threads per client for this, one blocked reading the socket and one writing broadcasts. Here it's one task and no locks around the socket.

Nicknames live in a HashMap behind a std::sync::Mutex shared through Arc. The lock is never held across an .await, so a plain Mutex is correct and cheaper than Tokio's async mutex. Each broadcast carries the sender's id, so users don't receive their own messages back.

To test the server, main starts it on a random free port and connects three scripted clients over real TCP. Each client waits for every line it expects before the script continues, so the conversation is deterministic even though everything runs concurrently. At the end, each client's transcript is printed.

Rust

use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{self, AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast::{self, error::RecvError};

// ---------- protocol.rs (step 1) ----------

#[derive(Debug, PartialEq)]
enum Command<'a> {
    Say(&'a str),
    Nick(&'a str),
    List,
    Quit,
    Invalid(String),
}

fn parse(line: &str) -> Command<'_> {
    let line = line.trim();
    let Some(command) = line.strip_prefix('/') else {
        return Command::Say(line);
    };
    let (name, argument) = command.split_once(' ').map_or((command, ""), |(n, a)| (n, a.trim()));
    match (name, argument) {
        ("nick", "") => Command::Invalid(String::from("usage: /nick <name>")),
        ("nick", nick) if nick.contains(' ') || nick.len() > 16 => {
            Command::Invalid(String::from("names are one word, at most 16 characters"))
        }
        ("nick", nick) => Command::Nick(nick),
        ("list", _) => Command::List,
        ("quit", _) => Command::Quit,
        (other, _) => Command::Invalid(format!("unknown command /{other}")),
    }
}

// ---------- server.rs ----------

#[derive(Clone, Debug)]
struct Event {
    from: usize, // connection id
    line: String,
}

struct Shared {
    names: Mutex<HashMap<usize, String>>,
    events: broadcast::Sender<Event>,
    next_id: AtomicUsize,
}

impl Shared {
    fn announce(&self, from: usize, line: String) {
        let _ = self.events.send(Event { from, line }); // Err only means nobody is listening
    }
}

async fn serve(listener: TcpListener, shared: Arc<Shared>) {
    loop {
        let Ok((stream, _addr)) = listener.accept().await else { continue };
        let shared = Arc::clone(&shared);
        tokio::spawn(async move {
            if let Err(error) = handle(stream, shared).await {
                eprintln!("connection error: {error}");
            }
        });
    }
}

async fn handle(stream: TcpStream, shared: Arc<Shared>) -> io::Result<()> {
    let id = shared.next_id.fetch_add(1, Ordering::Relaxed) + 1;
    let mut name = format!("guest-{id}");
    shared.names.lock().unwrap().insert(id, name.clone());

    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    let mut events = shared.events.subscribe(); // subscribe before announcing, so no event is missed

    writer.write_all(format!("welcome! you are {name}. commands: /nick <name>, /list, /quit\n").as_bytes()).await?;
    shared.announce(id, format!("{name} joined"));

    loop {
        tokio::select! {
            line = lines.next_line() => {
                let Some(line) = line? else { break }; // the client closed the connection
                let reply = match parse(&line) {
                    Command::Say("") => None,
                    Command::Say(text) => {
                        shared.announce(id, format!("{name}: {text}"));
                        None
                    }
                    Command::Nick(new_name) => {
                        let mut names = shared.names.lock().unwrap();
                        if names.values().any(|n| n == new_name) {
                            Some(format!("error: the name {new_name} is taken"))
                        } else {
                            names.insert(id, new_name.to_string());
                            drop(names); // release the lock before doing anything else
                            shared.announce(id, format!("{name} is now {new_name}"));
                            name = new_name.to_string();
                            Some(format!("you are now {name}"))
                        }
                    }
                    Command::List => {
                        let mut names: Vec<String> = shared.names.lock().unwrap().values().cloned().collect();
                        names.sort();
                        Some(format!("users: {}", names.join(", ")))
                    }
                    Command::Quit => break,
                    Command::Invalid(message) => Some(format!("error: {message}")),
                };
                if let Some(reply) = reply {
                    writer.write_all(format!("{reply}\n").as_bytes()).await?;
                }
            }
            event = events.recv() => match event {
                Ok(event) if event.from != id => writer.write_all(format!("{}\n", event.line).as_bytes()).await?,
                Ok(_) => {} // our own message
                Err(RecvError::Lagged(missed)) => {
                    writer.write_all(format!("(you missed {missed} messages)\n").as_bytes()).await?;
                }
                Err(RecvError::Closed) => break,
            }
        }
    }

    shared.names.lock().unwrap().remove(&id);
    shared.announce(id, format!("{name} left"));
    Ok(())
}

// ---------- scripted test clients ----------

struct Client {
    who: &'static str,
    lines: Lines<BufReader<OwnedReadHalf>>,
    writer: OwnedWriteHalf,
    transcript: Vec<String>,
}

impl Client {
    async fn connect(port: u16, who: &'static str) -> Client {
        let stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
        let (reader, writer) = stream.into_split();
        let mut client = Client { who, lines: BufReader::new(reader).lines(), writer, transcript: Vec::new() };
        client.expect().await; // the welcome line
        client
    }

    async fn send(&mut self, line: &str) {
        self.transcript.push(format!("> {line}"));
        self.writer.write_all(format!("{line}\n").as_bytes()).await.unwrap();
    }

    /// Waits for the next line from the server (at most one second).
    async fn expect(&mut self) {
        let line = tokio::time::timeout(Duration::from_secs(1), self.lines.next_line()).await;
        match line {
            Ok(Ok(Some(line))) => self.transcript.push(line),
            Ok(Ok(None)) => self.transcript.push(String::from("(disconnected)")),
            _ => panic!("{} expected a line but got nothing", self.who),
        }
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); // port 0: any free port
    let port = listener.local_addr().unwrap().port();
    let (events, _) = broadcast::channel(64);
    let shared = Arc::new(Shared { names: Mutex::new(HashMap::new()), events, next_id: AtomicUsize::new(0) });
    tokio::spawn(serve(listener, shared));

    let mut ana = Client::connect(port, "ana").await;
    ana.send("/nick ana").await;
    ana.expect().await;

    let mut ben = Client::connect(port, "ben").await;
    ana.expect().await; // guest-2 joined
    ben.send("/nick ben").await;
    ben.expect().await;
    ana.expect().await; // guest-2 is now ben

    ana.send("hi ben").await;
    ben.expect().await;
    ben.send("hi ana!").await;
    ana.expect().await;

    ben.send("/nick ana").await;
    ben.expect().await; // taken
    ben.send("/list").await;
    ben.expect().await;

    let mut cy = Client::connect(port, "cy").await;
    ana.expect().await;
    ben.expect().await;
    cy.send("hello, all").await;
    ana.expect().await;
    ben.expect().await;

    ben.send("/quit").await;
    ben.expect().await; // the server closes ben's connection
    ana.expect().await;
    cy.expect().await;

    cy.send("/shout hey").await;
    cy.expect().await;
    cy.send("/list").await;
    cy.expect().await;

    for client in [&ana, &ben, &cy] {
        println!("--- {}'s terminal", client.who);
        for line in &client.transcript {
            println!("{line}");
        }
    }
}

Output

--- ana's terminal
welcome! you are guest-1. commands: /nick <name>, /list, /quit
> /nick ana
you are now ana
guest-2 joined
guest-2 is now ben
> hi ben
ben: hi ana!
guest-3 joined
guest-3: hello, all
ben left
--- ben's terminal
welcome! you are guest-2. commands: /nick <name>, /list, /quit
> /nick ben
you are now ben
ana: hi ben
> hi ana!
> /nick ana
error: the name ana is taken
> /list
users: ana, ben
guest-3 joined
guest-3: hello, all
> /quit
(disconnected)
--- cy's terminal
welcome! you are guest-3. commands: /nick <name>, /list, /quit
> hello, all
ben left
> /shout hey
error: unknown command /shout
> /list
users: ana, guest-3

The server code is about 100 lines and handles any number of clients on a handful of threads. Everything the tasks share goes through two things: the broadcast channel for messages and the Mutex<HashMap> for names.

Ideas for extending it

Rooms: /join rust with one broadcast channel per room in a HashMap<String, broadcast::Sender<Event>>.
Private messages with /msg ben hello: keep an mpsc::Sender per connection in the shared map and send directly to one user.
Graceful shutdown on Ctrl-C with tokio::signal::ctrl_c() and a watch channel, as in the Key-Value Server project.
A WebSocket front end with axum's ws feature, so browsers can join; the connection loop stays the same shape.
Rate limiting per user, to stop one client from flooding everyone.

Summary

Keep the protocol in pure functions (parse) and test them without sockets.
broadcast delivers every message to every subscriber; a slow subscriber gets Lagged(n) instead of blocking the others.
One task per connection with select! over "the user typed a line" and "a broadcast arrived" replaces the two threads per client a blocking design would need.
Shared state that's only touched briefly and never across .await belongs in a std::sync::Mutex inside an Arc.
End-to-end tests can run real TCP clients against a server on port 0, waiting for each expected line so the result is deterministic.

Next: REST API with axum and SQLite.