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

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

Related explainers

typescript

type AsyncMethod = (...args: any[]) => Promise<any>;
 
function LogExecutionTime(thresholdMs = 0) {
  return function <T extends AsyncMethod>(

A method decorator that times async calls

decorators higher-order-functions async

Advanced 9 steps

rust

use axum::{
    body::Body,
    http::{header, HeaderValue, StatusCode, Uri},
    response::{IntoResponse, Response},

Serving embedded static files in Axum

static-assets embedding http-headers

Intermediate 7 steps

javascript

class InfiniteScroll {
  constructor(sentinel, { loadMore, root = null, rootMargin = '200px' } = {}) {
    this.sentinel = sentinel;
    this.loadMore = loadMore;

Infinite scroll with IntersectionObserver

intersectionobserver pagination async

Intermediate 10 steps

ruby

class Debouncer
  def initialize(delay:)
    @delay = delay
    @mutex = Mutex.new

A thread-safe debouncer in Ruby

concurrency debouncing mutex

Advanced 5 steps

ruby

require "net/http"
require "json"
 
class UrlFetcher

A thread-pool URL fetcher in Ruby

concurrency thread-pool queues

Intermediate 8 steps

rust

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;

A round-robin worker pool in Rust

concurrency thread-pool channels

Intermediate 7 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