rust 48 lines · 8 steps

Streaming live price ticks with SSE in Axum

An Axum handler turns a broadcast channel into a Server-Sent Events stream that pushes JSON price updates to every connected client.

Explained by highlit
Axum Advanced server-sent-events broadcast-channel streaming async shared-state
1use axum::{
2 response::sse::{Event, KeepAlive, Sse},
3 routing::get,
4 Router,
5};
6use futures::stream::Stream;
7use serde::Serialize;
8use std::{convert::Infallible, sync::Arc, time::Duration};
9use tokio::sync::broadcast;
10use tokio_stream::{wrappers::BroadcastStream, StreamExt};
11 
12#[derive(Clone, Serialize)]
13struct PriceTick {
14 symbol: String,
15 price: f64,
16 change: f64,
17}
18 
19#[derive(Clone)]
20struct AppState {
21 ticks: broadcast::Sender<PriceTick>,
22}
23 
24pub fn routes(ticks: broadcast::Sender<PriceTick>) -> Router {
25 Router::new()
26 .route("/prices/stream", get(stream_prices))
27 .with_state(Arc::new(AppState { ticks }))
28}
29 
30async fn stream_prices(
31 axum::extract::State(state): axum::extract::State<Arc<AppState>>,
32) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
33 let stream = BroadcastStream::new(state.ticks.subscribe()).filter_map(|tick| {
34 let tick = tick.ok()?;
35 let event = Event::default()
36 .event("tick")
37 .id(tick.symbol.clone())
38 .json_data(&tick)
39 .ok()?;
40 Some(Ok(event))
41 });
42 
43 Sse::new(stream).keep_alive(
44 KeepAlive::new()
45 .interval(Duration::from_secs(15))
46 .text("keep-alive"),
47 )
48}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A broadcast channel fans one producer out to many subscribers, making it a natural backbone for live SSE feeds.
  2. 2Mapping a stream into Result<Event, _> lets Axum handle the SSE wire format while you focus on payload shape.
  3. 3Keep-alive comments prevent idle connections and proxies from dropping a long-lived event stream.

Related explainers

rust 
use axum::{extract::{Path, State}, http::StatusCode, Json};
use dashmap::DashMap;
use serde::Serialize;
use std::sync::Arc;

Request coalescing in an Axum handler

caching concurrency request-coalescing
Advanced 8 steps
python 
from functools import wraps
import asyncio
 
from fastapi import APIRouter, FastAPI, Request

Per-route request timeouts in FastAPI

decorators async timeouts
Intermediate 6 steps
rust 
use std::net::Ipv4Addr;
use std::str::FromStr;
 
#[derive(Debug, Clone, Copy)]

Parsing and matching IPv4 CIDR ranges in Rust

bitwise-operations parsing error-handling
Intermediate 8 steps
rust 
use std::sync::OnceLock;
 
static CRC_TABLE: OnceLock<[u32; 256]> = OnceLock::new();
 

Table-driven CRC32 with lazy init in Rust

checksums lazy-initialization lookup-tables
Intermediate 8 steps
go 
func UploadChunk(c *gin.Context) {
	uploadID := c.Param("uploadID")
	if !validUploadID.MatchString(uploadID) {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid upload id"})

Resumable chunked uploads in Gin

file-upload content-range streaming
Advanced 9 steps
rust 
use axum::{
    body::Body,
    extract::Path,
    http::{header, HeaderMap, HeaderValue, StatusCode},

HTTP range requests for video streaming in Axum

http-range-requests streaming async-io
Advanced 8 steps

Share this explainer

Here's the card — post it anywhere.

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