Workshop 5: Streams and Protocols

Each project starts with a problem and requirements, then grows in two stages. Every stage is a complete program. The last stage is the full program, followed by a table of its parts and extension tasks.

The projects use only earlier lessons: Streams, Bytes and Framing, Codecs with tokio-util, mpsc Channels, oneshot, broadcast, and watch, Semaphore, Notify, Barrier, and OnceCell, TCP Clients and Servers, and Time: sleep, timeout, and interval.

The Rust blocks in this lesson are flagged rust,ignore. The example checker does not run them. Their output was derived by reading the code, and each program is written so that the output does not depend on scheduling. Servers bind 127.0.0.1:0, and clients run in the same program.

Project 21: Length-delimited chat protocol

A chat server and its clients exchange typed messages over TCP. TCP has no message boundaries, so each message is one length-prefixed frame. The frame body is a small binary format built with bytes.

Requirements:

Three messages: Join(name), Say(text), and Chat { from, text }.
A frame body is a one-byte tag followed by strings, each a two-byte big-endian length and UTF-8 bytes.
Decoding reports bad input as an io::Error and never panics.
The server broadcasts every chat message to every connected client, including the sender.

Stage 1: The message format

encode writes with BufMut methods. decode reads with Buf methods and checks remaining before every read, because Buf getters panic when the buffer is too short.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};
use std::io;

#[derive(Debug, Clone, PartialEq)]
enum Msg {
    Join(String),
    Say(String),
    Chat { from: String, text: String },
}

fn bad(reason: &str) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, reason.to_string())
}

fn put_str(buf: &mut BytesMut, s: &str) {
    buf.put_u16(s.len() as u16);
    buf.put_slice(s.as_bytes());
}

fn get_str(buf: &mut Bytes) -> io::Result<String> {
    if buf.remaining() < 2 {
        return Err(bad("truncated string"));
    }
    let len = buf.get_u16() as usize;
    if buf.remaining() < len {
        return Err(bad("truncated string"));
    }
    String::from_utf8(buf.split_to(len).to_vec()).map_err(|_| bad("invalid utf-8"))
}

fn encode(msg: &Msg) -> Bytes {
    let mut buf = BytesMut::new();
    match msg {
        Msg::Join(name) => { buf.put_u8(1); put_str(&mut buf, name); }
        Msg::Say(text) => { buf.put_u8(2); put_str(&mut buf, text); }
        Msg::Chat { from, text } => { buf.put_u8(3); put_str(&mut buf, from); put_str(&mut buf, text); }
    }
    buf.freeze()
}

fn decode(mut buf: Bytes) -> io::Result<Msg> {
    if buf.remaining() < 1 {
        return Err(bad("empty frame"));
    }
    match buf.get_u8() {
        1 => Ok(Msg::Join(get_str(&mut buf)?)),
        2 => Ok(Msg::Say(get_str(&mut buf)?)),
        3 => Ok(Msg::Chat { from: get_str(&mut buf)?, text: get_str(&mut buf)? }),
        tag => Err(bad(&format!("unknown tag {tag}"))),
    }
}

fn main() {
    let messages = [
        Msg::Join("alice".into()),
        Msg::Say("hi".into()),
        Msg::Chat { from: "alice".into(), text: "hi".into() },
    ];
    for msg in &messages {
        let bytes = encode(msg);
        println!("{msg:?} -> {} bytes -> {:?}", bytes.len(), decode(bytes.clone()).unwrap());
    }
    println!("bad tag: {}", decode(Bytes::from_static(&[9])).unwrap_err());
    println!("truncated: {}", decode(Bytes::from_static(&[1, 0, 5, b'a'])).unwrap_err());
}

Output

Join("alice") -> 8 bytes -> Join("alice")
Say("hi") -> 5 bytes -> Say("hi")
Chat { from: "alice", text: "hi" } -> 12 bytes -> Chat { from: "alice", text: "hi" }
bad tag: unknown tag 9
truncated: truncated string

Join("alice") is 1 tag byte, 2 length bytes, and 5 name bytes. The last line declares five bytes but supplies one.

Stage 2: Framed connections and a broadcast room

Framed<TcpStream, LengthDelimitedCodec> turns the socket into a Stream of frames and a Sink of frames, each with a four-byte length prefix. The codec yields BytesMut, so freeze converts a frame to Bytes for decode. Each server connection subscribes to a broadcast channel and uses select! to handle a frame from its client or a message from the room.

The client side runs in main and drives the steps one at a time. Each step waits for the reply it expects before the next starts, which fixes the order. This is the full program.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};
use futures::{SinkExt, StreamExt};
use std::io;
use std::net::SocketAddr;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio_util::codec::{Framed, LengthDelimitedCodec};

#[derive(Debug, Clone, PartialEq)]
enum Msg {
    Join(String),
    Say(String),
    Chat { from: String, text: String },
}

fn bad(reason: &str) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, reason.to_string())
}

fn put_str(buf: &mut BytesMut, s: &str) {
    buf.put_u16(s.len() as u16);
    buf.put_slice(s.as_bytes());
}

fn get_str(buf: &mut Bytes) -> io::Result<String> {
    if buf.remaining() < 2 {
        return Err(bad("truncated string"));
    }
    let len = buf.get_u16() as usize;
    if buf.remaining() < len {
        return Err(bad("truncated string"));
    }
    String::from_utf8(buf.split_to(len).to_vec()).map_err(|_| bad("invalid utf-8"))
}

fn encode(msg: &Msg) -> Bytes {
    let mut buf = BytesMut::new();
    match msg {
        Msg::Join(name) => { buf.put_u8(1); put_str(&mut buf, name); }
        Msg::Say(text) => { buf.put_u8(2); put_str(&mut buf, text); }
        Msg::Chat { from, text } => { buf.put_u8(3); put_str(&mut buf, from); put_str(&mut buf, text); }
    }
    buf.freeze()
}

fn decode(mut buf: Bytes) -> io::Result<Msg> {
    if buf.remaining() < 1 {
        return Err(bad("empty frame"));
    }
    match buf.get_u8() {
        1 => Ok(Msg::Join(get_str(&mut buf)?)),
        2 => Ok(Msg::Say(get_str(&mut buf)?)),
        3 => Ok(Msg::Chat { from: get_str(&mut buf)?, text: get_str(&mut buf)? }),
        tag => Err(bad(&format!("unknown tag {tag}"))),
    }
}

async fn serve_connection(stream: TcpStream, room: broadcast::Sender<Msg>) {
    let mut framed = Framed::new(stream, LengthDelimitedCodec::new());
    let mut updates = room.subscribe();
    let mut name = String::new();
    loop {
        tokio::select! {
            frame = framed.next() => {
                let Some(Ok(frame)) = frame else { break };
                match decode(frame.freeze()) {
                    Ok(Msg::Join(who)) => {
                        let text = format!("{who} joined");
                        name = who;
                        let _ = room.send(Msg::Chat { from: "server".into(), text });
                    }
                    Ok(Msg::Say(text)) => {
                        let _ = room.send(Msg::Chat { from: name.clone(), text });
                    }
                    _ => break,
                }
            }
            Ok(msg) = updates.recv() => {
                if framed.send(encode(&msg)).await.is_err() {
                    break;
                }
            }
        }
    }
}

async fn serve(listener: TcpListener) {
    let (room, _) = broadcast::channel(32);
    loop {
        let (stream, _) = listener.accept().await.unwrap();
        tokio::spawn(serve_connection(stream, room.clone()));
    }
}

struct Client { framed: Framed<TcpStream, LengthDelimitedCodec> }

impl Client {
    async fn connect(addr: SocketAddr) -> Client {
        let stream = TcpStream::connect(addr).await.unwrap();
        Client { framed: Framed::new(stream, LengthDelimitedCodec::new()) }
    }
    async fn send(&mut self, msg: Msg) {
        self.framed.send(encode(&msg)).await.unwrap();
    }
    async fn recv(&mut self) -> Msg {
        decode(self.framed.next().await.unwrap().unwrap().freeze()).unwrap()
    }
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(serve(listener));

    let mut alice = Client::connect(addr).await;
    alice.send(Msg::Join("alice".into())).await;
    println!("alice got: {:?}", alice.recv().await);

    let mut bob = Client::connect(addr).await;
    bob.send(Msg::Join("bob".into())).await;
    println!("bob got: {:?}", bob.recv().await);
    println!("alice got: {:?}", alice.recv().await);

    alice.send(Msg::Say("hi".into())).await;
    println!("alice got: {:?}", alice.recv().await);
    println!("bob got: {:?}", bob.recv().await);

    bob.send(Msg::Say("hello".into())).await;
    println!("alice got: {:?}", alice.recv().await);
    println!("bob got: {:?}", bob.recv().await);
}

Output

alice got: Chat { from: "server", text: "alice joined" }
bob got: Chat { from: "server", text: "bob joined" }
alice got: Chat { from: "server", text: "bob joined" }
alice got: Chat { from: "alice", text: "hi" }
bob got: Chat { from: "alice", text: "hi" }
alice got: Chat { from: "bob", text: "hello" }
bob got: Chat { from: "bob", text: "hello" }

Each connection subscribes before it reads its first frame. So when a client has received its own join notice, the room already delivers to it, and later messages cannot be missed. The codec removes all length handling from the program: partial frames and several frames in one read are its job.

Part	Role
`encode` / `decode`	binary message format with checked reads
`LengthDelimitedCodec`	four-byte length prefix, default 8 MiB maximum frame
`Framed`	one connection as `Stream` plus `Sink`
`broadcast` channel	delivers each chat message to every connection
`select!` in the handler	serves the client's frames and the room's messages at once

Extension tasks:

Add a Leave(name) message and broadcast it when a connection ends.
Set max_frame_length to 1024 on the codec and show that an oversized frame ends the connection.
Add a Whisper { to, text } message. Keep a map from name to mpsc::Sender<Msg> in place of the broadcast channel.

Project 22: Streaming ETL pipeline

A pipeline reads raw text records, drops the malformed ones, enriches each record with a slow lookup, groups records into batches, and loads each batch with a slow writer. Each stage pulls from the one before it, so a slow writer slows the whole pipeline.

Requirements:

The source produces 12 lines of the form id,amount. Lines with ids 5 and 10 are malformed.
Up to three enrichments run at once, and results keep their input order.
Records are grouped into batches of at most 4, or after 500 ms of waiting.
The loader takes 200 ms per batch. A channel of size 1 connects it to the pipeline.
The program reports how many lines were bad and how many records were loaded.

Stage 1: Parse and filter

map parses each line into a Result, and collect gathers all results. The two malformed lines become Err values that the program counts.

Rust

use futures::{StreamExt, stream};

#[derive(Debug)]
struct Rec { id: u32, amount: u32 }

fn parse(line: &str) -> Result<Rec, String> {
    let parts: Vec<&str> = line.split(',').collect();
    match parts.as_slice() {
        [id, amount] => Ok(Rec {
            id: id.parse().map_err(|_| format!("bad id in {line:?}"))?,
            amount: amount.parse().map_err(|_| format!("bad amount in {line:?}"))?,
        }),
        _ => Err(format!("wrong field count in {line:?}")),
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let lines = ["1,10", "2,x", "3,25", "4,", "5,40", "7,abc,9", "8,30"];
    let parsed: Vec<Result<Rec, String>> = stream::iter(lines).map(parse).collect().await;
    for item in &parsed {
        match item {
            Ok(rec) => println!("ok: {rec:?}"),
            Err(e) => println!("skipped: {e}"),
        }
    }
}

Output

ok: Rec { id: 1, amount: 10 }
skipped: bad amount in "2,x"
ok: Rec { id: 3, amount: 25 }
skipped: bad amount in "4,"
ok: Rec { id: 5, amount: 40 }
skipped: wrong field count in "7,abc,9"
ok: Rec { id: 8, amount: 30 }

Stage 2: Enrich, batch, and load with backpressure

The pipeline is one chain of stream adapters:

filter_map drops bad lines and counts them.
map followed by buffered(3) runs up to three enrichments at once and yields results in input order. buffer_unordered would yield them as they finish.
chunks_timeout(4, 500 ms) comes from tokio_stream. Both futures::StreamExt and tokio_stream::StreamExt define methods named next and map, so the program imports the futures trait and calls chunks_timeout by its full path.
A for loop sends each batch into a channel of capacity 1. send waits while the channel is full, so the slow loader stops the loop from pulling more records. Streams are pull-based, so the source runs only as fast as this loop asks.

The enrichment latency depends on the id, so the work finishes out of order internally, and buffered restores the order. The batches are [1,2,3,4], [6,7,8,9], and [11,12]. The last one is flushed when the stream ends.

Rust

use futures::{StreamExt, stream};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[derive(Debug)]
struct Rec { id: u32, amount: u32 }

#[derive(Debug)]
struct Enriched { id: u32, total: u32 }

fn parse(line: &str) -> Result<Rec, String> {
    let (id, amount) = line.split_once(',').ok_or("no comma")?;
    Ok(Rec {
        id: id.parse().map_err(|_| "bad id".to_string())?,
        amount: amount.parse().map_err(|_| "bad amount".to_string())?,
    })
}

async fn enrich(rec: Rec) -> Enriched {
    sleep(Duration::from_millis((rec.id as u64 % 3 + 1) * 20)).await;
    Enriched { id: rec.id, total: rec.amount + rec.amount / 10 }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let bad = Arc::new(AtomicUsize::new(0));
    let bad_in_stream = bad.clone();

    let enriched = stream::iter(1..=12u32)
        .map(|id| if id % 5 == 0 { format!("{id},oops") } else { format!("{id},{}", id * 10) })
        .filter_map(move |line| {
            let bad = bad_in_stream.clone();
            async move {
                match parse(&line) {
                    Ok(rec) => Some(rec),
                    Err(_) => { bad.fetch_add(1, SeqCst); None }
                }
            }
        })
        .map(enrich)
        .buffered(3);
    let batches = tokio_stream::StreamExt::chunks_timeout(enriched, 4, Duration::from_millis(500));
    tokio::pin!(batches);

    let (tx, mut rx) = mpsc::channel::<Vec<Enriched>>(1);
    let loader = tokio::spawn(async move {
        let (mut count, mut rows) = (0, 0);
        while let Some(batch) = rx.recv().await {
            sleep(Duration::from_millis(200)).await;
            count += 1;
            rows += batch.len();
            let ids: Vec<u32> = batch.iter().map(|r| r.id).collect();
            let sum: u32 = batch.iter().map(|r| r.total).sum();
            println!("batch {count}: ids={ids:?} sum={sum}");
        }
        (count, rows)
    });

    while let Some(batch) = batches.next().await {
        tx.send(batch).await.unwrap();
    }
    drop(tx);

    let (count, rows) = loader.await.unwrap();
    println!("bad lines: {}", bad.load(SeqCst));
    println!("loaded {rows} records in {count} batches");
}

Output

batch 1: ids=[1, 2, 3, 4] sum=110
batch 2: ids=[6, 7, 8, 9] sum=330
batch 3: ids=[11, 12] sum=253
bad lines: 2
loaded 10 records in 3 batches

The sum of total is amount + amount / 10 for each record. For batch 1 that is 11 + 22 + 33 + 44 = 110. Dropping tx closes the channel, which ends the loader loop after the last batch.

Part	Role
`stream::iter`	source of 12 ids
`filter_map`	drops malformed lines and counts them
`buffered(3)`	three concurrent enrichments, results in input order
`chunks_timeout(4, 500 ms)`	groups records into batches of at most 4
`mpsc::channel(1)`	bounded hand-off to the loader, so a slow loader slows the loop
loader task	writes one batch at a time

Extension tasks:

Replace buffered(3) with buffer_unordered(3) and sort each batch by id before loading.
Send the bad lines to a second bounded channel that a dead-letter task drains.
Lower the timeout to 30 ms and add a sleep to the source. Show that partial batches appear.

Project 23: Mini-Redis with TTL expiry and pub/sub

A small in-memory server speaks a line-based text protocol. It stores keys with optional expiry times, removes expired keys in the background, and lets clients subscribe to channels.

Requirements:

Commands: SET key value [PX millis], GET key, DEL key, DBSIZE, SUBSCRIBE channel, PUBLISH channel message. Values have no spaces.
Replies: +OK, $value, $nil, :number, -ERR reason, and !channel message for pushed messages.
A custom codec implements Decoder for commands and Encoder for replies.
An expiry task removes keys at their deadline. It sleeps until the earliest deadline and wakes early through a Notify when a new earlier deadline is set.
GET never returns an expired value, even if the expiry task has not run yet.

Stage 1: The codec and a key-value store

The codec reads up to the first \n, splits that line off the buffer, and parses it. If there is no newline yet, decode returns Ok(None) and the framing layer reads more bytes. A malformed command becomes Cmd::Bad, so the connection stays open and the client gets an error reply. The client side uses LinesCodec.

Rust

use bytes::BytesMut;
use futures::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::io;
use std::sync::{Arc, Mutex};
use tokio::net::{TcpListener, TcpStream};
use tokio_util::codec::{Decoder, Encoder, Framed, LinesCodec};

enum Cmd { Set(String, String), Get(String), Del(String), Bad(String) }
enum Resp { Ok, Value(String), Nil, Int(i64), Err(String) }

struct Codec;

fn parse(line: &str) -> Cmd {
    let parts: Vec<&str> = line.split_whitespace().collect();
    match parts.as_slice() {
        ["SET", k, v] => Cmd::Set(k.to_string(), v.to_string()),
        ["GET", k] => Cmd::Get(k.to_string()),
        ["DEL", k] => Cmd::Del(k.to_string()),
        _ => Cmd::Bad(format!("unknown command '{line}'")),
    }
}

impl Decoder for Codec {
    type Item = Cmd;
    type Error = io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<Cmd>> {
        let Some(end) = src.iter().position(|b| *b == b'\n') else {
            return Ok(None);
        };
        let line = src.split_to(end + 1);
        let text = std::str::from_utf8(&line[..end])
            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid utf-8"))?;
        Ok(Some(parse(text.trim_end_matches('\r'))))
    }
}

impl Encoder<Resp> for Codec {
    type Error = io::Error;

    fn encode(&mut self, resp: Resp, dst: &mut BytesMut) -> io::Result<()> {
        let text = match resp {
            Resp::Ok => "+OK".to_string(),
            Resp::Value(v) => format!("${v}"),
            Resp::Nil => "$nil".to_string(),
            Resp::Int(n) => format!(":{n}"),
            Resp::Err(e) => format!("-ERR {e}"),
        };
        dst.extend_from_slice(text.as_bytes());
        dst.extend_from_slice(b"\n");
        Ok(())
    }
}

type Db = Arc<Mutex<HashMap<String, String>>>;

async fn conn(stream: TcpStream, db: Db) {
    let mut framed = Framed::new(stream, Codec);
    while let Some(Ok(cmd)) = framed.next().await {
        let resp = match cmd {
            Cmd::Set(k, v) => { db.lock().unwrap().insert(k, v); Resp::Ok }
            Cmd::Get(k) => db.lock().unwrap().get(&k).cloned().map_or(Resp::Nil, Resp::Value),
            Cmd::Del(k) => Resp::Int(db.lock().unwrap().remove(&k).is_some() as i64),
            Cmd::Bad(reason) => Resp::Err(reason),
        };
        if framed.send(resp).await.is_err() {
            break;
        }
    }
}

async fn ask(client: &mut Framed<TcpStream, LinesCodec>, line: &str) -> String {
    client.send(line).await.unwrap();
    client.next().await.unwrap().unwrap()
}

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let db: Db = Arc::default();
    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            tokio::spawn(conn(stream, db.clone()));
        }
    });

    let stream = TcpStream::connect(addr).await.unwrap();
    let mut client = Framed::new(stream, LinesCodec::new());
    for line in ["SET k v", "GET k", "GET missing", "DEL k", "DEL k", "BAD"] {
        println!("{line} -> {}", ask(&mut client, line).await);
    }
}

Output

SET k v -> +OK
GET k -> $v
GET missing -> $nil
DEL k -> :1
DEL k -> :0
BAD -> -ERR unknown command 'BAD'

The custom codec is the only place that knows the wire format. conn works with Cmd and Resp values.

Stage 2: TTL, background expiry, and pub/sub

The database now keeps two structures under one lock: a map from key to (value, deadline), and a BTreeSet<(Instant, key)> of deadlines ordered by time, so the earliest deadline is the first element.

set replaces any old deadline for the key, then calls notify_one on a Notify. Notify stores one permit when nobody is waiting, so a wake sent while the expiry task is busy is not lost.
The expiry task calls purge, which removes every expired key and returns the next deadline. It then waits for either sleep_until(deadline) or notified(). With no deadline it waits for notified() alone.
get compares the deadline with the current time, so an expired key is never returned even before the task runs.
SUBSCRIBE creates a broadcast::Receiver for the channel and spawns a small task that forwards each message into a per-connection mpsc channel. The connection loop uses select! on its frames and on that channel, then writes !channel message replies. PUBLISH replies with the number of receivers that got the message.

The program runs under the paused clock. sleep in main moves time forward, and the expiry task runs at each deadline on the way. This is the full program.

Rust

use bytes::BytesMut;
use futures::{SinkExt, StreamExt};
use std::collections::{BTreeSet, HashMap};
use std::io;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{Notify, broadcast, mpsc};
use tokio::time::{Instant, sleep, sleep_until};
use tokio_util::codec::{Decoder, Encoder, Framed, LinesCodec};

enum Cmd {
    Set(String, String, Option<Duration>),
    Get(String),
    Del(String),
    Size,
    Subscribe(String),
    Publish(String, String),
    Bad(String),
}

enum Resp { Ok, Value(String), Nil, Int(i64), Push(String, String), Err(String) }

struct Codec;

fn parse(line: &str) -> Cmd {
    let parts: Vec<&str> = line.split_whitespace().collect();
    match parts.as_slice() {
        ["SET", k, v] => Cmd::Set(k.to_string(), v.to_string(), None),
        ["SET", k, v, "PX", ms] => match ms.parse() {
            Ok(ms) => Cmd::Set(k.to_string(), v.to_string(), Some(Duration::from_millis(ms))),
            Err(_) => Cmd::Bad("bad PX value".to_string()),
        },
        ["GET", k] => Cmd::Get(k.to_string()),
        ["DEL", k] => Cmd::Del(k.to_string()),
        ["DBSIZE"] => Cmd::Size,
        ["SUBSCRIBE", c] => Cmd::Subscribe(c.to_string()),
        ["PUBLISH", c, m] => Cmd::Publish(c.to_string(), m.to_string()),
        _ => Cmd::Bad(format!("unknown command '{line}'")),
    }
}

impl Decoder for Codec {
    type Item = Cmd;
    type Error = io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<Cmd>> {
        let Some(end) = src.iter().position(|b| *b == b'\n') else {
            return Ok(None);
        };
        let line = src.split_to(end + 1);
        let text = std::str::from_utf8(&line[..end])
            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid utf-8"))?;
        Ok(Some(parse(text.trim_end_matches('\r'))))
    }
}

impl Encoder<Resp> for Codec {
    type Error = io::Error;

    fn encode(&mut self, resp: Resp, dst: &mut BytesMut) -> io::Result<()> {
        let text = match resp {
            Resp::Ok => "+OK".to_string(),
            Resp::Value(v) => format!("${v}"),
            Resp::Nil => "$nil".to_string(),
            Resp::Int(n) => format!(":{n}"),
            Resp::Push(channel, message) => format!("!{channel} {message}"),
            Resp::Err(e) => format!("-ERR {e}"),
        };
        dst.extend_from_slice(text.as_bytes());
        dst.extend_from_slice(b"\n");
        Ok(())
    }
}

#[derive(Default)]
struct State {
    data: HashMap<String, (String, Option<Instant>)>,
    deadlines: BTreeSet<(Instant, String)>,
}

#[derive(Default)]
struct Db {
    state: Mutex<State>,
    wake: Notify,
    channels: Mutex<HashMap<String, broadcast::Sender<String>>>,
}

impl Db {
    fn set(&self, key: String, value: String, ttl: Option<Duration>) {
        let deadline = ttl.map(|t| Instant::now() + t);
        let mut st = self.state.lock().unwrap();
        if let Some((_, Some(old))) = st.data.insert(key.clone(), (value, deadline)) {
            st.deadlines.remove(&(old, key.clone()));
        }
        if let Some(d) = deadline {
            st.deadlines.insert((d, key));
        }
        drop(st);
        self.wake.notify_one();
    }

    fn get(&self, key: &str) -> Option<String> {
        match self.state.lock().unwrap().data.get(key) {
            Some((_, Some(deadline))) if *deadline <= Instant::now() => None,
            Some((value, _)) => Some(value.clone()),
            None => None,
        }
    }

    fn del(&self, key: &str) -> bool {
        let mut st = self.state.lock().unwrap();
        match st.data.remove(key) {
            Some((_, Some(d))) => { st.deadlines.remove(&(d, key.to_string())); true }
            Some(_) => true,
            None => false,
        }
    }

    fn size(&self) -> usize {
        self.state.lock().unwrap().data.len()
    }

    fn purge(&self) -> Option<Instant> {
        let mut st = self.state.lock().unwrap();
        let now = Instant::now();
        while let Some((deadline, key)) = st.deadlines.first().cloned() {
            if deadline > now {
                return Some(deadline);
            }
            st.deadlines.pop_first();
            st.data.remove(&key);
        }
        None
    }

    fn channel(&self, name: &str) -> broadcast::Sender<String> {
        let mut channels = self.channels.lock().unwrap();
        channels.entry(name.to_string()).or_insert_with(|| broadcast::channel(16).0).clone()
    }
}

async fn expirer(db: Arc<Db>) {
    loop {
        match db.purge() {
            Some(when) => {
                tokio::select! {
                    _ = sleep_until(when) => {}
                    _ = db.wake.notified() => {}
                }
            }
            None => db.wake.notified().await,
        }
    }
}

async fn conn(stream: TcpStream, db: Arc<Db>) {
    let mut framed = Framed::new(stream, Codec);
    let (push_tx, mut push_rx) = mpsc::channel::<Resp>(16);
    loop {
        tokio::select! {
            frame = framed.next() => {
                let Some(Ok(cmd)) = frame else { break };
                let resp = match cmd {
                    Cmd::Set(k, v, ttl) => { db.set(k, v, ttl); Resp::Ok }
                    Cmd::Get(k) => db.get(&k).map_or(Resp::Nil, Resp::Value),
                    Cmd::Del(k) => Resp::Int(db.del(&k) as i64),
                    Cmd::Size => Resp::Int(db.size() as i64),
                    Cmd::Subscribe(channel) => {
                        let mut rx = db.channel(&channel).subscribe();
                        let out = push_tx.clone();
                        tokio::spawn(async move {
                            while let Ok(message) = rx.recv().await {
                                if out.send(Resp::Push(channel.clone(), message)).await.is_err() {
                                    break;
                                }
                            }
                        });
                        Resp::Ok
                    }
                    Cmd::Publish(channel, message) => {
                        Resp::Int(db.channel(&channel).send(message).unwrap_or(0) as i64)
                    }
                    Cmd::Bad(reason) => Resp::Err(reason),
                };
                if framed.send(resp).await.is_err() {
                    break;
                }
            }
            Some(push) = push_rx.recv() => {
                if framed.send(push).await.is_err() {
                    break;
                }
            }
        }
    }
}

async fn ask(client: &mut Framed<TcpStream, LinesCodec>, line: &str) -> String {
    client.send(line).await.unwrap();
    client.next().await.unwrap().unwrap()
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let db = Arc::new(Db::default());
    tokio::spawn(expirer(db.clone()));
    tokio::spawn({
        let db = db.clone();
        async move {
            loop {
                let (stream, _) = listener.accept().await.unwrap();
                tokio::spawn(conn(stream, db.clone()));
            }
        }
    });

    let mut a = Framed::new(TcpStream::connect(addr).await.unwrap(), LinesCodec::new());
    let mut b = Framed::new(TcpStream::connect(addr).await.unwrap(), LinesCodec::new());

    for line in ["SET a 1 PX 100", "SET b 2 PX 300", "SET c 3", "GET a", "DBSIZE"] {
        println!("{line} -> {}", ask(&mut a, line).await);
    }
    sleep(Duration::from_millis(150)).await;
    for line in ["GET a", "DBSIZE"] {
        println!("after 150 ms: {line} -> {}", ask(&mut a, line).await);
    }
    sleep(Duration::from_millis(200)).await;
    for line in ["GET b", "DBSIZE"] {
        println!("after 350 ms: {line} -> {}", ask(&mut a, line).await);
    }
    println!("BAD -> {}", ask(&mut a, "BAD").await);

    println!("SUBSCRIBE news -> {}", ask(&mut b, "SUBSCRIBE news").await);
    println!("PUBLISH news hello -> {}", ask(&mut a, "PUBLISH news hello").await);
    println!("subscriber got: {}", b.next().await.unwrap().unwrap());
    println!("PUBLISH other x -> {}", ask(&mut a, "PUBLISH other x").await);
}

Output

SET a 1 PX 100 -> +OK
SET b 2 PX 300 -> +OK
SET c 3 -> +OK
GET a -> $1
DBSIZE -> :3
after 150 ms: GET a -> $nil
after 150 ms: DBSIZE -> :2
after 350 ms: GET b -> $nil
after 350 ms: DBSIZE -> :1
BAD -> -ERR unknown command 'BAD'
SUBSCRIBE news -> +OK
PUBLISH news hello -> :1
subscriber got: !news hello
PUBLISH other x -> :0

DBSIZE falls from 3 to 2 to 1 because the expiry task removes keys at their deadlines. Lazy expiry in get alone would leave the entries in the map. The first DBSIZE after 150 ms would still be 3. PUBLISH other x reaches no subscriber, so it replies :0.

Part	Role
`Codec`	`Decoder` for commands and `Encoder` for replies; the only code that knows the wire format
`Db.state`	key-value map plus a `BTreeSet` of deadlines ordered by time
`expirer` with `Notify`	sleeps to the earliest deadline, wakes early when `set` adds one
lazy check in `get`	correct reads between a deadline and the next purge
`broadcast` per channel	fan-out to subscribers; `send` returns the receiver count
forwarder task and `mpsc`	moves broadcast messages into the connection's own loop

Extension tasks:

Add UNSUBSCRIBE channel by keeping the forwarder's JoinHandle per channel and aborting it.
Add a maximum line length to the decoder and reply -ERR line too long instead of buffering without limit.
Add TTL key, which replies with the remaining milliseconds or -1 when the key has no deadline.

Interview points

TCP delivers a byte stream. A protocol needs framing, such as a length prefix or a delimiter.
Buf getters panic on short input. Decoders check remaining first and return an error.
Framed combines a Stream of decoded items and a Sink of encoded items on one connection. A decode error ends the stream.
buffered(n) keeps input order and buffer_unordered(n) does not. Both run at most n futures at once.
chunks_timeout(n, d) emits a batch when it has n items, when d has passed since the batch started, or when the stream ends.
Streams are pull-based. A bounded channel between stages makes a slow stage slow the earlier ones.
Notify::notify_one stores a permit when no task waits, so a wake that arrives before notified() is not lost.
A TTL store needs both lazy expiry, for correct reads, and a background purge, for memory.
A BTreeSet ordered by deadline gives the next expiry in one lookup.
Returning a malformed-command value from a decoder keeps the connection open. Returning an error ends it.

Summary

Project 21 encodes typed messages with bytes, frames them with LengthDelimitedCodec, and fans them out with broadcast.
Project 22 chains filter_map, buffered, and chunks_timeout, and connects the last stage through a bounded channel.
Project 23 implements Decoder and Encoder for a text protocol, expires keys with a Notify-driven task, and forwards broadcast messages to subscribers.