Build a Chat Server in Rust With Tokio Broadcast Channels

A chat server is the smallest program that needs real concurrent design: many readers, many writers and one shared stream of events. It is a good project after the TCP basics in async networking with Tokio. This version fits in one file of about 60 lines, and it handles slow and dropped clients properly. I compiled and ran everything here on Rust 1.99 (edition 2024) with Tokio 1.47.1 on an Apple M1 Max, including a scripted three client session whose output appears below.

[dependencies]
tokio = { version = "=1.47.1", features = ["full"] }

The problem is a fan out. A line typed by one client has to reach every other client, and the clients are independent: each reads at its own speed, and any can disappear at any time. Two designs come to mind first. Keep a list of client senders and loop over it for every message, or give each client a task that talks to a shared room. Tokio's broadcast channel is the second design, packaged, and it makes the three hard parts (storage, slow readers, cleanup) properties of the channel instead of code you write.

Design

• One broadcast::Sender<(SocketAddr, String)> for the whole room. The address says who sent a message so a client does not hear itself.
• One task per client that selects between "a line arrived from this client's socket" and "a message arrived from the room".
• A bounded buffer of 64 messages, so memory stays flat when a client stops reading.

A broadcast channel stores each message once in a ring buffer of the given capacity. Every receiver has its own read position into that ring. Sending writes one slot; receiving reads the slot at the receiver's position and clones the value out. That is why a message costs one stored copy plus one clone per receiver, not one stored copy per receiver, and it is why the type has to implement Clone.

The server

use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;

type Message = (SocketAddr, String);

async fn client(stream: TcpStream, addr: SocketAddr, tx: broadcast::Sender<Message>) {
    let mut rx = tx.subscribe();
    let (read_half, mut write_half) = stream.into_split();
    let mut lines = BufReader::new(read_half).lines();

    let _ = tx.send((addr, format!("{addr} joined")));

    loop {
        tokio::select! {
            line = lines.next_line() => {
                match line {
                    Ok(Some(text)) => {
                        let _ = tx.send((addr, format!("{addr}: {text}")));
                    }
                    _ => break,
                }
            }
            msg = rx.recv() => {
                match msg {
                    Ok((from, text)) => {
                        if from != addr
                            && (write_half.write_all(text.as_bytes()).await.is_err()
                                || write_half.write_all(b"\n").await.is_err())
                        {
                            break;
                        }
                    }
                    Err(RecvError::Lagged(missed)) => {
                        let note = format!("(you missed {missed} messages)\n");
                        if write_half.write_all(note.as_bytes()).await.is_err() {
                            break;
                        }
                    }
                    Err(RecvError::Closed) => break,
                }
            }
        }
    }

    let _ = tx.send((addr, format!("{addr} left")));
}

pub async fn serve(listener: TcpListener) -> std::io::Result<()> {
    let (tx, _) = broadcast::channel::<Message>(64);
    loop {
        let (stream, addr) = listener.accept().await?;
        tokio::spawn(client(stream, addr, tx.clone()));
    }
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:7001").await?;
    serve(listener).await
}

Take it in order. type Message pairs the sender's socket address with the finished line to display. client is the per connection task; serve accepts connections and spawns one of them for each. The listener is a parameter of serve instead of being created inside it so the demo below can bind port 0 and ask the OS for a free port. main binds the real port 7001.

Inside client, tx.subscribe() comes first, and stream.into_split() splits the socket into independent read and write halves so one can sit in a select! branch while the other is written to. BufReader::new(read_half).lines() turns the read half into a stream of lines. The room is told "joined" after subscribing, so the join message reaches every other client; the new client's own copy is dropped by the address check.

The loop is one tokio::select! with two branches. The first waits for lines.next_line(): a line becomes a broadcast message tagged with this client's address, and None or an error means the peer is gone, so the loop breaks. The second waits for rx.recv(). A message from someone else is written to the socket followed by a newline. A write error ends the loop. Lagged(missed) is the interesting arm: this client fell more than 64 messages behind, the channel overwrote what it had not read, and the client is told how many it missed before it carries on with the oldest message still held. Closed means every sender was dropped, which cannot happen while serve holds one but is handled so the loop has a defined end.

To use it, start the server and open three terminals with nc 127.0.0.1 7001. Lines typed in one appear in the others, prefixed with the sender's address. Because that needs three terminals, here is the same thing scripted:

nc 127.0.0.1 7001

#[path = "c1.rs"]
#[allow(dead_code)]
mod c1;
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::sleep;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    tokio::spawn(c1::serve(listener));

    let mut alice = TcpStream::connect(addr).await?;
    let (bob_r, mut bob_w) = TcpStream::connect(addr).await?.into_split();
    sleep(Duration::from_millis(50)).await;

    alice.write_all(b"hello from alice\n").await?;
    bob_w.write_all(b"hi alice\n").await?;
    sleep(Duration::from_millis(50)).await;
    drop(bob_w);
    drop(bob_r);
    sleep(Duration::from_millis(50)).await;

    // Alice's socket holds everything the room sent her, nothing she sent herself.
    alice.shutdown().await?;
    let mut lines = BufReader::new(alice).lines();
    while let Ok(Some(l)) = tokio::time::timeout(Duration::from_millis(200), lines.next_line()).await.unwrap_or(Ok(None)) {
        println!("alice saw: {l}");
    }
    Ok(())
}

The demo binds an ephemeral port, runs the server in a spawned task, connects Alice and Bob, has each say a line, then closes Bob. It reads what arrived at Alice's socket:

alice saw: 127.0.0.1:64228 joined
alice saw: 127.0.0.1:64228: hi alice
alice saw: 127.0.0.1:64228 left

Alice sees that Bob joined, Bob's line, and that Bob left. She does not see "hello from alice", her own line, because the room drops messages whose sender address is hers. The prefix is Bob's address; port 64228 was whatever the OS handed out on that run. Bob's own copy of Alice's line went to his socket too, but I only printed Alice's side.

Why it is shaped this way

Subscribe before announcing. A receiver only sees messages sent after it subscribes. Creating the receiver first means nothing the client should see can slip past between "connected" and "listening". It matters more once you add history replay, where the order of subscribe and replay decides whether a message can be seen twice or lost.

Both branches of the select! are cancel safe. select! drops the branch that did not win, and a future dropped halfway through can lose data. lines.next_line() and rx.recv() can both be dropped and restarted without losing a line or a message (cancellation safety). The write is inside the branch body, not part of the future being raced, so it is never cancelled. The cost is that while a write is in progress, this client is not reading its socket or the room. That is acceptable, and it is the mechanism of the next point.

Slow clients lag; they do not leak. A client whose socket is full blocks in write_all, stops calling recv, and falls behind in the ring. When it returns to recv it learns how many messages it missed. The sender never waits, which differs from a bounded mpsc, where the producer would block until the slowest consumer caught up (bounded channels and backpressure). For chat that is the right trade: one frozen phone should not freeze the room. For a job queue it would be wrong.

The lag behaviour is easy to see in isolation. This program sends ten numbers into a channel of capacity four before a receiver reads any:

use tokio::sync::broadcast::{self, error::RecvError};

#[tokio::main]
async fn main() {
    let (tx, mut slow) = broadcast::channel::<u32>(4);
    for i in 1..=10 {
        tx.send(i).unwrap();
    }
    loop {
        match slow.recv().await {
            Ok(v) => println!("got {v}"),
            Err(RecvError::Lagged(n)) => println!("lagged, missed {n}"),
            Err(RecvError::Closed) => break,
        }
        if slow.is_empty() { break; }
    }
}

lagged, missed 6
got 7
got 8
got 9
got 10

The first recv reports Lagged(6): six values were overwritten before it could read them. After that it receives 7, 8, 9 and 10, the four the ring still held. The lag is reported once, then reading resumes from the oldest surviving message. A client that wants lossless delivery has to be fast enough, or the capacity large enough; there is no third option.

Disconnects clean themselves up. When the task ends, rx is dropped, and dropping a receiver unsubscribes it. There is no list of clients to prune. tx.send returns an error only when there are no receivers at all, which is why the code ignores the result: an empty room is not a failure for a chat server.

A failing example: the message type must be Clone

The first thing that goes wrong when you adapt this is the message type. Say you replace the tuple with a struct:

use tokio::sync::broadcast;

#[derive(Debug)]
struct Line {
    from: u16,
    text: String,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = broadcast::channel::<Line>(16);
    tx.send(Line { from: 1, text: "hi".into() }).unwrap();
    let got = rx.recv().await.unwrap();
    println!("{} {}", got.from, got.text);
}

The compiler refuses, before anything runs:

error[E0277]: the trait bound `Line: Clone` is not satisfied
   --> src/bin/c3_fail.rs:11:45
    |
 11 |     let (tx, mut rx) = broadcast::channel::<Line>(16);
    |                                             ^^^^ the trait `Clone` is not implemented for `Line`
    |
note: required by a bound in `tokio::sync::broadcast::channel`
   --> tokio-1.47.1/src/sync/broadcast.rs:508:19
    |
508 | pub fn channel<T: Clone>(capacity: usize) -> (Sender<T>, Receiver<T>) {
    |                   ^^^^^ required by this bound in `channel`
help: consider annotating `Line` with `#[derive(Clone)]`
    |
  4 + #[derive(Clone)]
  5 | struct Line {
    |

I trimmed the second error, which says that recv is also unavailable for the same reason. The cause is the design described above: each receiver gets its own copy of every message, so the channel needs to clone it, and channel declares T: Clone as a bound. The suggested #[derive(Clone)] is the fix. If the clone is expensive, put the payload behind an Arc and send that: cloning an Arc<str> is one atomic increment, not a copy of the text. For a chat line that is premature; for a stream of kilobyte payloads it is the first optimisation to make.

Following one message through the server

Suppose Alice types "hello" and presses return. Her client task is parked inside the select!, polling lines.next_line() and rx.recv() together. The bytes arrive on her socket, the Tokio reactor wakes her task, and the select! polls both branches again. The line branch is ready, so its arm runs: it formats "addr: hello" and calls tx.send. send is synchronous and never waits; it writes the value into the next slot of the ring, bumps the tail and wakes every receiver that was waiting for a new message.

Each of those receivers belongs to another client task, parked in its own select!. They wake, poll rx.recv(), clone the message out of the ring and reach the arm that checks the sender address. For everyone but Alice the address differs, so each writes the text and a newline to its own socket. Alice's task wakes too, because her own receiver is subscribed, and reads the message it just sent; the address matches, so it writes nothing. That wasted wake up is the price of filtering at the receiver. If rooms get large, filtering at the sender is cheaper: keep a separate channel per recipient and skip yourself, which brings back the list you were avoiding.

Nothing in this path takes a lock that is held across a wait. send locks the ring briefly to claim a slot, and each recv locks it briefly to read one. The sender cannot be blocked by a slow reader, because the ring overwrites the oldest slot when full; the reader pays instead, with a Lagged error. Everything that can wait is an await in a client task, and each client task waits only for itself.

Slow sockets, and where the backpressure goes

A subtle point: the broadcast channel has no backpressure, but TCP does. When a client stops reading, the kernel's send buffer for that connection fills, and write_all on that socket stops completing. The client's task is now parked inside the match arm, not in the select!, so it stops pulling from rx and from the socket. The room keeps moving, the ring keeps overwriting, and when the socket drains the task returns to recv and receives Lagged(n) with whatever it missed. The chain is: slow peer, full TCP buffer, parked write, lagging receiver, a notice. Each link is a mechanism you did not have to write.

The weak point is the same chain read the other way. A client that never reads at all keeps its task and its socket forever, and with each lag notice it also gets a fresh write_all that cannot finish. Put a timeout around the write, using tokio::time::timeout, and drop the client when it fires. Five seconds is a defensible start for chat; measure what your clients do and adjust.

Choosing the capacity

The capacity of 64 in the server is a guess, and the right way to guess is arithmetic. The ring holds at most that many messages, whatever the number of clients, so the memory ceiling is capacity times the average message size, plus per receiver bookkeeping. At 64 messages of about 100 bytes, the ring is about 6 KB, which is nothing. Raising it to 4,096 costs about 400 KB and buys tolerance for a client that stalls for 4,096 messages instead of 64. In a room that sees ten messages a second, 64 is six seconds of grace, and 4,096 is almost seven minutes.

So the question is how long a stall you want to forgive. A mobile client that goes through a tunnel and comes back inside ten seconds should not be told it missed messages; one that vanishes for an hour will be told, and the right response to Lagged then is not to print a notice but to resync from storage. Set the capacity from the grace period and the peak message rate, not from habit. The lag count you log is the feedback: if it is never non-zero, the buffer is larger than you need, and if it fires daily for real users, it is too small.

Testing the server without a terminal

The demo above is also the pattern for a real test, and one choice in it matters. It binds port 0, so the operating system picks a free port and parallel test runs never collide on 7001. It then reads the address back with local_addr. The pauses are crude. Sleeping 50 milliseconds and hoping the server caught up is the thing that makes network tests flaky on a loaded CI machine. A sturdier version reads from each client socket with a timeout until the expected line shows up, and fails with the lines it did see. The structure stays the same, and the assertion moves from "print what arrived" to "the set of lines Alice read equals the set I sent".

Test the three behaviours that the design promises: a sender does not hear itself, a disconnect produces a "left" message, and a client that falls behind receives the lag notice. The first and second are in the demo's output. The third needs a capacity small enough to overflow cheaply, which is a reason to make the capacity an argument of serve instead of a literal inside it.

What fan out costs

"Clones every message to every receiver" invites a measurement. I sent 100,000 strings of the form message 12345 through a channel with a capacity of 1,024, with a multi thread runtime of four workers, and varied the number of receivers. The sender waits when more than 512 are queued so nobody lags; the test is about delivery cost.

use std::time::Instant;
use tokio::sync::broadcast;

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() {
    const MSGS: u32 = 100_000;
    for receivers in [1usize, 10, 100, 1000] {
        let (tx, _) = broadcast::channel::<String>(1024);
        let mut handles = Vec::new();
        for _ in 0..receivers {
            let mut rx = tx.subscribe();
            handles.push(tokio::spawn(async move {
                let mut seen = 0u32;
                let mut lagged = 0u64;
                loop {
                    match rx.recv().await {
                        Ok(_) => seen += 1,
                        Err(broadcast::error::RecvError::Lagged(n)) => lagged += n,
                        Err(_) => break,
                    }
                }
                (seen, lagged)
            }));
        }
        let t = Instant::now();
        for i in 0..MSGS {
            while tx.len() > 512 { tokio::task::yield_now().await; }
            tx.send(format!("message {i}")).unwrap();
        }
        drop(tx);
        let mut total_seen = 0u64;
        let mut total_lag = 0u64;
        for h in handles { let (s, l) = h.await.unwrap(); total_seen += s as u64; total_lag += l; }
        println!("{receivers:>5} receivers: {:>7.1?}  delivered {total_seen}  lagged {total_lag}", t.elapsed());
    }
}

    1 receivers:  30.7ms  delivered 100000  lagged 0
   10 receivers:  65.4ms  delivered 1000000  lagged 0
  100 receivers: 461.6ms  delivered 10000000  lagged 0
 1000 receivers:    4.5s  delivered 100000000  lagged 0

Divide the time by the number of deliveries. One receiver costs about 307 ns per delivery (30.7 ms for 100,000), because a lone receiver pays the full price of waking its task for nearly every message. Ten receivers cost 65 ns each, a hundred cost 46 ns and a thousand cost 45 ns (4.5 seconds for 100 million deliveries). Past a handful of receivers the work grows linearly at roughly 45 nanoseconds per delivery, so a 1,000 client room spends about 45 microseconds of CPU on each line typed. A single room with a few thousand idle clients is fine at chat typing speeds. At a few thousand messages a second to a thousand clients, you are spending a core on cloning. Past that, shard clients across several runtimes or send a pre-serialised Arc<[u8]> to avoid per receiver allocations. These numbers measure the channel; real clients add the socket write, which dominates.

Rooms

A single room is a global variable with extra steps. Real use wants named rooms, and the extension is small: a map from room name to its own Sender, behind a lock. A client joins by subscribing to the room's sender and leaves by dropping its receiver.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;

#[derive(Clone, Default)]
pub struct Rooms {
    inner: Arc<Mutex<HashMap<String, broadcast::Sender<String>>>>,
}

impl Rooms {
    pub fn join(&self, name: &str) -> (broadcast::Sender<String>, broadcast::Receiver<String>) {
        let mut map = self.inner.lock().unwrap();
        let tx = map
            .entry(name.to_owned())
            .or_insert_with(|| broadcast::channel(64).0)
            .clone();
        let rx = tx.subscribe();
        (tx, rx)
    }

    // Called after a client leaves: a room nobody listens to is dropped.
    pub fn prune(&self) {
        self.inner.lock().unwrap().retain(|_, tx| tx.receiver_count() > 0);
    }

    pub fn len(&self) -> usize {
        self.inner.lock().unwrap().len()
    }
}

#[tokio::main]
async fn main() {
    let rooms = Rooms::default();
    let (tx_a, mut rx_a) = rooms.join("rust");
    let (_tx_b, rx_b) = rooms.join("rust");
    let (_tx_c, rx_c) = rooms.join("async");
    tx_a.send("hello rust".into()).unwrap();
    println!("rust listener got: {}", rx_a.recv().await.unwrap());
    println!("rooms: {}", rooms.len());
    drop(rx_c);
    drop((_tx_c, rx_b));
    rooms.prune();
    println!("rooms after async emptied: {}", rooms.len());
}

rust listener got: hello rust
rooms: 2
rooms after async emptied: 1

The lock is a std::sync::Mutex, held only for the map operation and never across an await, which is the rule that matters here (mutex guard across await). Two clients joined "rust", the first sent a line, and the receiver in the same room got it. After the only "async" subscriber left, prune found the room had receiver_count() == 0 and removed it; the "rust" room, with a listener still attached, stayed. Without pruning, the map grows with every room name ever created, which is a memory leak an attacker can trigger by joining random names. For shared state patterns beyond this one, see sharing state between tasks.

Limits of this design

• Lines are unbounded. lines() buffers until it sees a newline, so a client that sends gigabytes without one makes the server allocate until it is killed. Cap the line length with a codec or wrap the reader in take before shipping this anywhere.
• A slow client is dropped to fewer messages, not disconnected. A client that is permanently slow lags forever and wastes a task. Count lag events and disconnect after a threshold.
• No authentication or identity. The address is the name. Behind a NAT several users share one address, and reconnecting changes the port.
• Messages are lost on restart. The ring buffer lives in memory. A chat that needs history needs storage.
• One process. The broadcast channel does not span machines. Multiple server instances need a message bus such as Redis pub/sub or NATS between them.

Broadcast is also the wrong tool when each message must reach exactly one consumer, or when no message may be dropped. In those cases a bounded mpsc or a work queue gives the guarantee, at the price of making the producer wait (Tokio channels compared lays out which channel gives which promise).

Improvements to try

1. Nicknames. Read the first line as a name and put it in the message instead of the address.
2. History. Keep the last 50 messages in a ring and send them to a client on join, after subscribing and before entering the loop, with the ordering caveat above.
3. Graceful shutdown. Add a cancellation token branch to the select and tell clients the server is going away (graceful shutdown).
4. A lag policy. Disconnect clients who lag more than a few times in a minute.
5. A real wire format. Length prefixed frames or JSON lines instead of raw text, so a message can carry its type.

For the web version of the same idea, where the clients are browsers and the transport is a WebSocket, see WebSockets with broadcast in Axum. The channel logic is identical; only the socket changes.