Server-Sent Events with Axum

18 min read

Quick Overview

Server-Sent Events (SSE) is a simple, HTTP-based protocol for pushing a one-way stream of text events from server to browser over a single long-lived connection — exactly what you reach for to power live dashboards, notification feeds, progress bars, or token-by-token LLM output. In Axum you model an SSE endpoint as an ordinary handler that returns axum::response::Sse wrapping an async Stream of events; this page shows how that Stream abstraction replaces the imperative res.write() loop you would write in Express.

Note: This page uses axum 0.8 (current stable 0.8.9) on the latest stable toolchain, Rust 1.96.0, 2024 edition. cargo new selects that edition automatically. SSE lives in axum::response::sse; you do not need a separate crate for the protocol itself, only a way to build a Stream (the futures-util, tokio-stream, or async-stream crates).

TypeScript/JavaScript Example

SSE is built into the browser as the EventSource API, and on the server it is just a normal HTTP response with Content-Type: text/event-stream that you keep open and write to. Here is a realistic Node endpoint (no framework, to make the wire format obvious) that pushes a tick event three times and then closes:

1
// sse.mts — Node v22, plain http server
2
import http from "node:http";
3

4
interface Tick {
5
  seq: number;
6
  message: string;
7
}
8

9
const server = http.createServer((req, res) => {
10
  if (req.url === "/events") {
11
    // The three headers that make a response an SSE stream.
12
    res.writeHead(200, {
13
      "Content-Type": "text/event-stream",
14
      "Cache-Control": "no-cache",
15
      Connection: "keep-alive",
16
    });
17

18
    let seq = 0;
19
    const timer = setInterval(() => {
20
      const tick: Tick = { seq, message: `tick ${seq}` };
21
      // The wire format: optional `event:` line, then `data:`, then a BLANK line.
22
      res.write(`event: tick\n`);
23
      res.write(`data: ${JSON.stringify(tick)}\n\n`);
24
      seq += 1;
25
      if (seq >= 3) {
26
        clearInterval(timer);
27
        res.end();
28
      }
29
    }, 150);
30

31
    // If the client disconnects, stop the timer so we don't leak it.
32
    req.on("close", () => clearInterval(timer));
33
  } else {
34
    res.writeHead(404).end();
35
  }
36
});
37

38
server.listen(3011, () => console.log("listening on http://127.0.0.1:3011"));

The browser side is tiny:

1
// client.ts — runs in the browser
2
const source = new EventSource("/events");
3
source.addEventListener("tick", (e: MessageEvent) => {
4
  const tick = JSON.parse(e.data) as { seq: number; message: string };
5
  console.log(tick.seq, tick.message);
6
});
7
source.onerror = () => console.log("connection lost; browser will auto-reconnect");

Things to notice in the Node version: you own the timer, you format every data: line and remember the trailing blank line, and you must clean up on req.on("close") or you leak the interval. Running it and connecting with curl -N yields:

1
event: tick
2
data: {"seq":0,"message":"tick 0"}
3

4
event: tick
5
data: {"seq":1,"message":"tick 1"}
6

7
event: tick
8
data: {"seq":2,"message":"tick 2"}

Rust Equivalent

In Axum you do not write to the socket imperatively. You return a Stream, and Axum’s Sse response type handles the headers, the data:/event: framing, the blank-line separators, keep-alive comments, and cleanup-on-disconnect for you.

Add the dependencies in a fresh project (cargo new sse-demo):

1
[dependencies]
2
axum = "0.8.9"
3
tokio = { version = "1.52.3", features = ["full"] }
4
tokio-stream = "0.1.18"
5
futures-util = "0.3.32"
6
serde = { version = "1.0.228", features = ["derive"] }
7
serde_json = "1.0.150"

1
use axum::{
2
    response::sse::{Event, KeepAlive, Sse},
3
    routing::get,
4
    Router,
5
};
6
use futures_util::stream::{self, Stream};
7
use serde::Serialize;
8
use std::{convert::Infallible, time::Duration};
9
use tokio_stream::StreamExt;
10

11
#[derive(Serialize)]
12
struct Tick {
13
    seq: u64,
14
    message: String,
15
}
16

17
// GET /events — a Server-Sent Events endpoint.
18
// The return type says: "a stream of `Event`s that can never fail".
19
async fn sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
20
    let stream = stream::iter(0..3)
21
        .map(|seq| {
22
            let tick = Tick { seq, message: format!("tick {seq}") };
23
            // `json_data` serializes with serde and sets the `data:` line.
24
            Ok(Event::default().event("tick").json_data(tick).unwrap())
25
        })
26
        // Space the events out in time, just like the Node `setInterval`.
27
        .throttle(Duration::from_millis(200));
28

29
    Sse::new(stream).keep_alive(KeepAlive::default())
30
}
31

32
#[tokio::main]
33
async fn main() {
34
    let app = Router::new().route("/events", get(sse_handler));
35
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3007").await.unwrap();
36
    println!("listening on http://{}", listener.local_addr().unwrap());
37
    axum::serve(listener, app).await.unwrap();
38
}

Run it (cargo run) and connect with curl -N http://127.0.0.1:3007/events. This is the real wire output (-N disables curl’s buffering so you see events as they arrive):

1
event: tick
2
data: {"seq":0,"message":"tick 0"}
3

4
event: tick
5
data: {"seq":1,"message":"tick 1"}
6

7
event: tick
8
data: {"seq":2,"message":"tick 2"}

Byte-for-byte the same protocol the Node server produced — but you never wrote a data: prefix, a blank-line separator, or a disconnect handler. The response headers are set for you; inspecting them with curl -D - shows:

1
HTTP/1.1 200 OK
2
content-type: text/event-stream
3
cache-control: no-cache
4
transfer-encoding: chunked

The same browser EventSource client from the TypeScript section talks to this endpoint without changes — SSE is a wire protocol, not a framework feature.

Detailed Explanation

The whole design hinges on one substitution: Node’s imperative res.write() loop becomes a declarative Stream. Let’s walk the Rust handler line by line and contrast each piece with the Node version.

The return type is the contract

1
async fn sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {

Sse<S> is an IntoResponse wrapper (see request-response.md for how IntoResponse turns values into HTTP responses). The type parameter S must be a Stream whose item is Result<Event, E> — note the Result. SSE streams are allowed to yield errors mid-stream (e.g. a database read fails), so every item is a Result. When the stream genuinely cannot fail, the error type is std::convert::Infallible — the empty type that says “this Err variant can never be constructed”.

A Stream is the async cousin of an Iterator: instead of next() -> Option<T>, it has poll_next() -> Poll<Option<T>>, so producing the next item can .await. Streams add the async dimension on top of the synchronous iterator model, and are covered in 11-async.

Building the stream

1
let stream = stream::iter(0..3)
2
    .map(|seq| { /* ... */ Ok(Event::default().event("tick").json_data(tick).unwrap()) })
3
    .throttle(Duration::from_millis(200));

stream::iter(0..3) lifts an ordinary iterator (0, 1, 2) into a Stream. This is the moral equivalent of the Node seq counter.
.map(...) transforms each number into a Result<Event, Infallible>. Event::default() builds an empty event; .event("tick") sets the event: field (the SSE “event name” the browser dispatches on); .json_data(tick) serializes the struct with serde and writes the data: line. We then wrap it in Ok(...).
.throttle(Duration::from_millis(200)) (from tokio_stream::StreamExt) ensures at least 200ms passes between items — the declarative replacement for setInterval. There is no timer handle to clear; when the connection closes, Axum drops the stream and the throttle timer dies with it.

The `Event` builder

Event is a builder for all five SSE fields. Each method returns Self, so they chain:

Method	SSE wire field	Purpose
`.data(s)` / `.json_data(v)?`	`data:`	The payload. `json_data` serializes with serde.
`.event(name)`	`event:`	Named event type the browser dispatches on (`addEventListener("tick", ...)`).
`.id(s)`	`id:`	Sets `lastEventId`; the browser echoes it as the `Last-Event-ID` header on reconnect.
`.retry(duration)`	`retry:`	Tells the browser how long to wait before reconnecting.
`.comment(s)`	`: ...`	A comment line (ignored by clients); used for keep-alive pings.

Note: json_data returns a Result because serialization can fail; in handlers prefer ? over .unwrap(), which is why the error-returning patterns below matter.

Keep-alive

1
Sse::new(stream).keep_alive(KeepAlive::default())

If your stream is idle for a long time, proxies and load balancers may kill the “dead” connection. KeepAlive periodically sends an SSE comment line (: plus optional text) that the browser ignores but that keeps the TCP connection and any intermediary timers alive. The default interval is 15 seconds; you can customize it (shown in the real-world example below). The Node version has no equivalent — you would have to schedule your own res.write(": ping\n\n").

Where the analogy breaks down

Unlike the Node example, there is no req.on("close") cleanup to write. Cancellation in Rust is structured: when the client disconnects, Axum stops polling the stream and drops it. Any resources the stream holds (timers, a broadcast::Receiver, a database cursor) are released by their Drop impls. You get correct cleanup for free precisely because you described what the stream is rather than imperatively driving it.

Key Differences

Concern	Express / Node	Axum
Mental model	Imperative: keep `res` open, `res.write(...)` repeatedly	Declarative: return a `Stream<Item = Result<Event, E>>`
Wire framing (`data:`, blank lines)	You format every line by hand	`Event` builder + `Sse` do it
Headers (`text/event-stream`, `no-cache`)	You set them via `res.writeHead`	Set automatically by the `Sse` response
Spacing events over time	`setInterval` + a timer handle	A stream combinator (`.throttle`) or an interval stream
Cleanup on disconnect	Manual `req.on("close")` handler	Automatic — the stream is dropped, `Drop` releases resources
Errors mid-stream	Throw / `res.destroy()`; ad hoc	First-class: each item is a `Result`, error type in the signature
Keep-alive pings	Schedule your own comment writes	`.keep_alive(KeepAlive::default())`
Fan-out to many clients	Track a `Set<res>` and loop-write	Each client subscribes to a `broadcast` channel

The deeper point: SSE in Express is a pattern you assemble; in Axum it is a type you return. The compiler enforces that you produce Events wrapped in Result, and the framework owns the protocol details.

Tip: SSE is one-way (server → client) and text-only over plain HTTP, with automatic browser reconnection built into EventSource. If you need bidirectional or binary traffic, use WebSockets instead — see websockets.md for the tradeoffs.

Common Pitfalls

Pitfall 1: forgetting that items must be `Result`, not bare `Event`

A TypeScript developer reasonably writes a stream that yields Event values directly:

1
use axum::{
2
    response::sse::{Event, Sse},
3
    routing::get,
4
    Router,
5
};
6
use futures_util::stream::{self, Stream};
7
use futures_util::StreamExt;
8

9
// does not compile (error E0271): items are `Event`, not `Result<Event, E>`.
10
async fn bad_handler() -> Sse<impl Stream<Item = Event>> {
11
    let stream = stream::iter(0..3).map(|i| Event::default().data(format!("{i}")));
12
    Sse::new(stream)
13
}
14

15
#[tokio::main]
16
async fn main() {
17
    let app: Router = Router::new().route("/events", get(bad_handler));
18
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3009").await.unwrap();
19
    axum::serve(listener, app).await.unwrap();
20
}

The real compiler error:

1
error[E0271]: expected `{closure@main.rs:11:41}` to return `Result<Event, _>`, but it returns `Event`
2
  --> src/main.rs:11:45
3
   |
4
11 |     let stream = stream::iter(0..3).map(|i| Event::default().data(format!("{i}")));
5
   |                                         --- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ expected `Result<Event, _>`, found `Event`
6
   |                                         |
7
   |                                         this closure
8
12 |     Sse::new(stream)
9
   |     ---------------- closure used here
10
   |
11
   = note: expected enum `Result<Event, _>`
12
            found struct `Event`

Fix: wrap each event in Ok(...) and annotate the error type, e.g. Result<Event, Infallible> when nothing can fail.

Pitfall 2: importing the wrong `StreamExt`

Both futures_util::StreamExt and tokio_stream::StreamExt exist, and they offer different combinators. .throttle(...) lives on tokio_stream::StreamExt; .enumerate() lives on futures_util::StreamExt. If you call a combinator without its trait in scope, the compiler complains that the trait bounds were not satisfied — but, helpfully, points you at the exact import you need:

1
// does not compile (error E0599): StreamExt is not in scope.
2
fn main() {
3
    let s = futures_util::stream::iter(0..3);
4
    let _e = s.enumerate();
5
}

1
error[E0599]: the method `enumerate` exists for struct `futures_util::stream::Iter<std::ops::Range<{integer}>>`, but its trait bounds were not satisfied
2
 --> src/main.rs:3:16
3
  |
4
3 |     let _e = s.enumerate();
5
  |                ^^^^^^^^^ method cannot be called due to unsatisfied trait bounds
6
  |
7
  = note: the following trait bounds were not satisfied:
8
          `futures_util::stream::Iter<std::ops::Range<{integer}>>: Iterator`
9
          which is required by `&mut futures_util::stream::Iter<std::ops::Range<{integer}>>: Iterator`
10
  = help: items from traits can only be used if the trait is in scope
11
help: trait `StreamExt` which provides `enumerate` is implemented but not in scope; perhaps you want to import it
12
  |
13
1 + use futures_util::StreamExt;
14
  |

Fix: read the help: line — it names the exact trait to import. Importing both StreamExt traits causes a method-ambiguity error for the methods they share (like map), so pick the one whose combinators you need, or fully-qualify the call.

Pitfall 3: blocking the async runtime to produce events

Putting a synchronous sleep or a blocking I/O call inside the stream’s closure stalls the entire worker thread, freezing every other request the runtime is handling:

1
// wrong: std::thread::sleep blocks the async worker thread.
2
.map(|i| {
3
    std::thread::sleep(std::time::Duration::from_secs(1)); // never do this in async
4
    Ok::<_, std::convert::Infallible>(axum::response::sse::Event::default().data(i.to_string()))
5
})

Fix: use async timing — .throttle(...), an IntervalStream, or tokio::time::sleep(...).await inside an async_stream::stream! { ... } block (shown next). See 11-async for why blocking calls poison cooperative runtimes.

Pitfall 4: buffering proxies hide your events

If you test through nginx (or a similar reverse proxy) and events arrive all at once at the end instead of trickling in, the proxy is buffering the response. SSE needs proxy_buffering off; (nginx) on the relevant location. This is not an Axum issue — Axum already sends Cache-Control: no-cache — but it surprises people. Locally, remember curl -N to disable curl’s own buffering. Deployment specifics live in deployment.md.

Best Practices

Always type the stream’s error explicitly. Use Infallible when nothing can fail (it documents intent and lets the compiler optimize), or a real error type when items come from fallible sources.
Prefer event.json_data(value)? over hand-formatting data:. It serializes with serde, escapes newlines correctly, and keeps your payload in one typed struct.
Set an event id when clients should resume. On reconnect the browser sends Last-Event-ID; read it (via the headers extractor — see extractors.md) to replay only what was missed.
Add a keep-alive. KeepAlive::default() is cheap insurance against idle-connection timeouts in proxies.
Fan out with a broadcast channel, not a shared Vec of senders. tokio::sync::broadcast gives each subscriber its own receiver and handles slow-consumer lag for you; combine it with shared State (see state-management.md).
Use async_stream::stream! when emission is inherently sequential/stateful (a countdown, a job that does work between yields). Use combinators (stream::iter(...).map(...).throttle(...)) when the source is already a collection or another stream.
Set Content-Type via Sse, never by hand. Returning Sse already sets it; manually adding the header risks duplicates.

Real-World Example

A live notification feed: clients open GET /events to subscribe, and any process can POST /messages to broadcast a message to every connected client. This is the canonical SSE production pattern — a tokio::sync::broadcast channel held in shared state, with each subscriber getting its own receiver wrapped as a stream.

Add tokio-stream’s sync feature for BroadcastStream (cargo add tokio-stream --features sync):

1
use axum::{
2
    extract::State,
3
    response::sse::{Event, KeepAlive, Sse},
4
    routing::{get, post},
5
    Json, Router,
6
};
7
use futures_util::stream::Stream;
8
use serde::{Deserialize, Serialize};
9
use std::{convert::Infallible, time::Duration};
10
use tokio::sync::broadcast;
11
use tokio_stream::wrappers::BroadcastStream;
12
use tokio_stream::StreamExt;
13

14
#[derive(Clone, Serialize)]
15
struct ChatMessage {
16
    user: String,
17
    text: String,
18
}
19

20
#[derive(Clone)]
21
struct AppState {
22
    // A broadcast channel: every subscriber gets a clone of every message.
23
    tx: broadcast::Sender<ChatMessage>,
24
}
25

26
#[derive(Deserialize)]
27
struct PostMessage {
28
    user: String,
29
    text: String,
30
}
31

32
// POST /messages — publish to all connected SSE clients.
33
async fn publish(State(state): State<AppState>, Json(body): Json<PostMessage>) -> &'static str {
34
    let msg = ChatMessage { user: body.user, text: body.text };
35
    // `send` errors only when there are zero receivers; we ignore that.
36
    let _ = state.tx.send(msg);
37
    "queued"
38
}
39

40
// GET /events — subscribe; the stream lives as long as the connection.
41
async fn subscribe(
42
    State(state): State<AppState>,
43
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
44
    let rx = state.tx.subscribe();
45
    let stream = BroadcastStream::new(rx).filter_map(|res| match res {
46
        Ok(msg) => Some(Ok(Event::default().json_data(msg).unwrap())),
47
        // A slow client that fell behind yields a `Lagged` error; skip it.
48
        Err(_lagged) => None,
49
    });
50

51
    Sse::new(stream).keep_alive(
52
        KeepAlive::new()
53
            .interval(Duration::from_secs(15))
54
            .text("keep-alive"),
55
    )
56
}
57

58
#[tokio::main]
59
async fn main() {
60
    let (tx, _rx) = broadcast::channel::<ChatMessage>(64);
61
    let state = AppState { tx };
62

63
    let app = Router::new()
64
        .route("/messages", post(publish))
65
        .route("/events", get(subscribe))
66
        .with_state(state);
67

68
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3008").await.unwrap();
69
    println!("listening on http://{}", listener.local_addr().unwrap());
70
    axum::serve(listener, app).await.unwrap();
71
}

Exercise it: start the server, subscribe in one terminal with curl -N http://127.0.0.1:3008/events, then post messages from another:

1
curl -X POST http://127.0.0.1:3008/messages \
2
  -H 'content-type: application/json' -d '{"user":"alice","text":"hello"}'
3
curl -X POST http://127.0.0.1:3008/messages \
4
  -H 'content-type: application/json' -d '{"user":"bob","text":"hi there"}'

The subscriber receives, in real time (verified output):

1
data: {"user":"alice","text":"hello"}
2

3
data: {"user":"bob","text":"hi there"}

When a subscriber’s curl is killed, its BroadcastStream is dropped, the underlying broadcast::Receiver is dropped, and the channel automatically stops tracking it — no bookkeeping, no leaked slot. In a real app, AppState would also hold a database pool (see state-management.md and ../17-database/README.md), and publish would persist the message before broadcasting.

Tip: broadcast::channel(64) sets the per-receiver buffer. If a client cannot keep up and falls more than 64 messages behind, it receives a Lagged error (which we map to “skip”). Size the buffer to your tolerance for dropped messages on slow clients.

Exercises

Exercise 1: A server clock

Difficulty: Beginner

Objective: Build the simplest possible time-driven SSE endpoint and watch it tick.

Instructions: Create a GET /clock handler that emits one event per second forever, where each event’s data: is the current count (0, 1, 2, …). Use tokio_stream::wrappers::IntervalStream over a tokio::time::interval plus a stream combinator. Add a default keep-alive. Connect with curl -N and confirm one event arrives per second.

Solution

1
use axum::{
2
    response::sse::{Event, KeepAlive, Sse},
3
    routing::get,
4
    Router,
5
};
6
use futures_util::stream::{Stream, StreamExt};
7
use std::{convert::Infallible, time::Duration};
8

9
async fn clock() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
10
    let stream = tokio_stream::wrappers::IntervalStream::new(
11
        tokio::time::interval(Duration::from_secs(1)),
12
    )
13
    .enumerate()
14
    .map(|(count, _instant)| Ok(Event::default().data(count.to_string())));
15

16
    Sse::new(stream).keep_alive(KeepAlive::default())
17
}
18

19
#[tokio::main]
20
async fn main() {
21
    let app = Router::new().route("/clock", get(clock));
22
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3012").await.unwrap();
23
    println!("listening on http://{}", listener.local_addr().unwrap());
24
    axum::serve(listener, app).await.unwrap();
25
}

curl -N http://127.0.0.1:3012/clock prints (one line pair per second):

1
data: 0
2

3
data: 1
4

5
data: 2

Note IntervalStream’s combinators here come from futures_util::StreamExt (enumerate, map). Import only that StreamExt to avoid the method-ambiguity error from Pitfall 2.

Exercise 2: A resumable metrics stream with event ids

Difficulty: Intermediate

Objective: Emit structured JSON events with monotonically increasing ids so a reconnecting client could resume.

Instructions: Create a GET /metrics handler that, once per second, emits a named metrics event whose data: is a JSON object { "cpu": ..., "memory": ..., "at": <RFC 3339 timestamp> }, and whose SSE id is the sequence number. Use serde for the struct and the chrono crate for the timestamp (cargo add chrono --features clock).

Solution

1
use axum::{
2
    response::sse::{Event, KeepAlive, Sse},
3
    routing::get,
4
    Router,
5
};
6
use chrono::Utc;
7
use futures_util::stream::{Stream, StreamExt};
8
use serde::Serialize;
9
use std::{convert::Infallible, time::Duration};
10

11
#[derive(Serialize)]
12
struct Reading {
13
    cpu: f64,
14
    memory: f64,
15
    at: String,
16
}
17

18
async fn metrics() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
19
    let stream = tokio_stream::wrappers::IntervalStream::new(
20
        tokio::time::interval(Duration::from_secs(1)),
21
    )
22
    .enumerate()
23
    .map(|(i, _instant)| {
24
        let reading = Reading {
25
            cpu: 12.5 + i as f64,
26
            memory: 40.0,
27
            at: Utc::now().to_rfc3339(),
28
        };
29
        Ok(Event::default()
30
            .id(i.to_string())
31
            .event("metrics")
32
            .json_data(reading)
33
            .unwrap())
34
    });
35

36
    Sse::new(stream).keep_alive(KeepAlive::default())
37
}
38

39
#[tokio::main]
40
async fn main() {
41
    let app = Router::new().route("/metrics", get(metrics));
42
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3012").await.unwrap();
43
    println!("listening on http://{}", listener.local_addr().unwrap());
44
    axum::serve(listener, app).await.unwrap();
45
}

curl -N http://127.0.0.1:3012/metrics prints (verified output):

1
id: 0
2
event: metrics
3
data: {"cpu":12.5,"memory":40.0,"at":"2026-06-01T12:07:52.873017+00:00"}
4

5
id: 1
6
event: metrics
7
data: {"cpu":13.5,"memory":40.0,"at":"2026-06-01T12:07:53.873344+00:00"}
8

9
id: 2
10
event: metrics
11
data: {"cpu":14.5,"memory":40.0,"at":"2026-06-01T12:07:54.876789+00:00"}

To make it truly resumable, add a headers: HeaderMap parameter (see extractors.md) and start the counter from the incoming Last-Event-ID header.

Exercise 3: A countdown with sequential async logic

Difficulty: Intermediate / Advanced

Objective: Use async_stream::stream! to emit events from imperative, stateful async code — the situation where combinators get awkward.

Instructions: Create a GET /launch handler that counts down T-minus 3, T-minus 2, T-minus 1 (one event every 150ms), then emits a final liftoff named event with payload Liftoff!, then ends the stream. Use the async_stream::stream! macro (cargo add async-stream) so you can write a normal for loop with .await and yield.

Solution

1
use axum::{
2
    response::sse::{Event, KeepAlive, Sse},
3
    routing::get,
4
    Router,
5
};
6
use async_stream::stream;
7
use futures_util::stream::Stream;
8
use std::{convert::Infallible, time::Duration};
9

10
async fn countdown() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
11
    let s = stream! {
12
        for n in (1..=3).rev() {
13
            tokio::time::sleep(Duration::from_millis(150)).await;
14
            yield Ok(Event::default().data(format!("T-minus {n}")));
15
        }
16
        yield Ok(Event::default().event("liftoff").data("Liftoff!"));
17
    };
18
    Sse::new(s).keep_alive(KeepAlive::default())
19
}
20

21
#[tokio::main]
22
async fn main() {
23
    let app = Router::new().route("/launch", get(countdown));
24
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3010").await.unwrap();
25
    println!("listening on http://{}", listener.local_addr().unwrap());
26
    axum::serve(listener, app).await.unwrap();
27
}

curl -N http://127.0.0.1:3010/launch prints (verified):

1
data: T-minus 3
2

3
data: T-minus 2
4

5
data: T-minus 1
6

7
event: liftoff
8
data: Liftoff!

The stream! macro turns this imperative async block into a Stream whose items are exactly what each yield produces — far more readable than chaining combinators when each step depends on the last. For fallible logic, use try_stream! and yield values directly while propagating errors with ?.

Server-Sent Events with Axum

Quick Overview

TypeScript/JavaScript Example

Rust Equivalent

Detailed Explanation

The return type is the contract

Building the stream

The Event builder

Keep-alive

Where the analogy breaks down

Key Differences

Common Pitfalls

Pitfall 1: forgetting that items must be Result, not bare Event

Pitfall 2: importing the wrong StreamExt

Pitfall 3: blocking the async runtime to produce events

Pitfall 4: buffering proxies hide your events

Best Practices

Real-World Example

Further Reading

Exercises

Exercise 1: A server clock

Exercise 2: A resumable metrics stream with event ids

Exercise 3: A countdown with sequential async logic

The `Event` builder

Pitfall 1: forgetting that items must be `Result`, not bare `Event`

Pitfall 2: importing the wrong `StreamExt`