rust 56 lines · 9 steps

Broadcasting live prices over SSE in Axum

A Tokio broadcast channel fans price ticks out to every SSE subscriber, turning a shared publisher into per-client event streams.

Explained by highlit
Axum Advanced server-sent-events broadcast-channel streams async pub-sub
1use axum::{
2 extract::State,
3 response::sse::{Event, KeepAlive, Sse},
4};
5use futures::stream::Stream;
6use serde::Serialize;
7use std::{convert::Infallible, sync::Arc, time::Duration};
8use tokio::sync::broadcast;
9use tokio_stream::{wrappers::BroadcastStream, StreamExt};
10 
11#[derive(Clone, Serialize)]
12pub struct PriceTick {
13 pub symbol: String,
14 pub price: f64,
15 pub ts: i64,
16}
17 
18#[derive(Clone)]
19pub struct AppState {
20 pub ticks: broadcast::Sender<PriceTick>,
21}
22 
23impl AppState {
24 pub fn new() -> Self {
25 let (ticks, _) = broadcast::channel(256);
26 Self { ticks }
27 }
28}
29 
30pub async fn stream_prices(
31 State(state): State<Arc<AppState>>,
32) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
33 let rx = state.ticks.subscribe();
34 
35 let stream = BroadcastStream::new(rx).filter_map(|res| match res {
36 Ok(tick) => Some(
37 Event::default()
38 .event("price")
39 .json_data(&tick)
40 .unwrap_or_else(|_| Event::default().comment("serialize error")),
41 ),
42 Err(_lagged) => None,
43 });
44 
45 let stream = stream.map(Ok);
46 
47 Sse::new(stream).keep_alive(
48 KeepAlive::new()
49 .interval(Duration::from_secs(15))
50 .text("keep-alive"),
51 )
52}
53 
54pub async fn publish_tick(State(state): State<Arc<AppState>>, tick: PriceTick) {
55 let _ = state.ticks.send(tick);
56}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A broadcast channel lets one producer fan messages out to many independent subscribers.
  2. 2Turning a receiver into a Stream lets Axum's Sse type push events to clients as they arrive.
  3. 3Handling the lagged case explicitly keeps a slow client from breaking the whole stream.

Related explainers

rust 
use axum::{
    body::{Body, Bytes},
    extract::Request,
    http::StatusCode,

Logging request bodies in Axum middleware

middleware streaming-body logging
Intermediate 8 steps
java 
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.regex.Pattern;

Ranking the most frequent words in Java

streams regex grouping
Intermediate 7 steps
rust 
use std::time::Duration;
 
use axum::{
    body::Body,

Turning Axum timeouts into 504 JSON

middleware timeouts error-handling
Intermediate 7 steps
rust 
pub struct Fibonacci {
    current: u64,
    next: u64,
}

A Fibonacci iterator in Rust

iterators traits lazy-evaluation
Intermediate 6 steps
rust 
use std::any::{Any, TypeId};
use std::collections::HashMap;
 
pub trait Event: Any {

A type-safe event bus in Rust

type-erasure trait-objects downcasting
Advanced 8 steps
rust 
use std::cmp::Ordering;
 
pub struct PrefixIndex {
    entries: Vec<String>,

Prefix search with binary partitioning in Rust

binary-search sorting case-insensitive
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Broadcasting live prices over SSE in Axum — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code