rust 66 lines · 8 steps

Long-polling with Axum and Tokio Notify

An Axum handler holds a request open until new messages arrive on a topic, or times out cleanly after 30 seconds.

Explained by highlit
Axum Advanced long-polling concurrency shared-state async timeout
1use std::sync::Arc;
2use std::time::Duration;
3 
4use axum::{
5 extract::{Path, State},
6 http::StatusCode,
7 Json,
8};
9use dashmap::DashMap;
10use serde::Serialize;
11use tokio::sync::Notify;
12use tokio::time::timeout;
13 
14#[derive(Clone)]
15struct Channel {
16 notify: Arc<Notify>,
17 messages: Arc<parking_lot::Mutex<Vec<Message>>>,
18}
19 
20#[derive(Clone, Serialize)]
21struct Message {
22 seq: u64,
23 body: String,
24}
25 
26#[derive(Clone, Default)]
27struct AppState {
28 channels: Arc<DashMap<String, Channel>>,
29}
30 
31async fn poll(
32 Path((topic, since)): Path<(String, u64)>,
33 State(state): State<AppState>,
34) -> Result<Json<Vec<Message>>, StatusCode> {
35 let channel = state
36 .channels
37 .entry(topic)
38 .or_insert_with(|| Channel {
39 notify: Arc::new(Notify::new()),
40 messages: Arc::new(parking_lot::Mutex::new(Vec::new())),
41 })
42 .clone();
43 
44 loop {
45 let notified = channel.notify.notified();
46 
47 {
48 let pending: Vec<Message> = channel
49 .messages
50 .lock()
51 .iter()
52 .filter(|m| m.seq > since)
53 .cloned()
54 .collect();
55 
56 if !pending.is_empty() {
57 return Ok(Json(pending));
58 }
59 }
60 
61 match timeout(Duration::from_secs(30), notified).await {
62 Ok(()) => continue,
63 Err(_) => return Ok(Json(Vec::new())),
64 }
65 }
66}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Long-polling trades many short requests for one held-open request that returns the moment data is ready.
  2. 2Registering the wake notification before checking for data avoids a race where a message slips in between the check and the wait.
  3. 3Wrapping an await in a timeout guarantees the handler releases the connection rather than hanging forever.

Related explainers

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
rust 
use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{self, BufRead, BufReader, BufWriter, Lines, Write};

K-way merge of sorted logs in Rust

binary-heap k-way-merge streaming-io
Intermediate 8 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
javascript 
import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks
Advanced 10 steps

Share this explainer

Here's the card — post it anywhere.

Long-polling with Axum and Tokio Notify — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code