rust 59 lines · 9 steps

Streaming NDJSON from Postgres in Axum

An Axum handler streams database rows to the client as newline-delimited JSON without buffering the whole result set.

Explained by highlit
Axum Advanced streaming backpressure async channels serialization
1use axum::{
2 body::Body,
3 extract::State,
4 http::{header, StatusCode},
5 response::{IntoResponse, Response},
6};
7use futures::Stream;
8use serde::Serialize;
9use tokio::sync::mpsc;
10use tokio_stream::wrappers::ReceiverStream;
11 
12#[derive(Serialize)]
13struct EventRecord {
14 id: i64,
15 kind: String,
16 payload: serde_json::Value,
17}
18 
19pub async fn stream_events(State(pool): State<sqlx::PgPool>) -> Response {
20 let (tx, rx) = mpsc::channel::<Result<String, std::io::Error>>(32);
21 
22 tokio::spawn(async move {
23 let mut cursor = sqlx::query_as::<_, (i64, String, serde_json::Value)>(
24 "SELECT id, kind, payload FROM events ORDER BY id",
25 )
26 .fetch(&pool);
27 
28 use futures::StreamExt;
29 while let Some(row) = cursor.next().await {
30 let line = match row {
31 Ok((id, kind, payload)) => {
32 let record = EventRecord { id, kind, payload };
33 match serde_json::to_string(&record) {
34 Ok(mut json) => {
35 json.push('\n');
36 Ok(json)
37 }
38 Err(err) => Err(std::io::Error::new(std::io::ErrorKind::Other, err)),
39 }
40 }
41 Err(err) => Err(std::io::Error::new(std::io::ErrorKind::Other, err)),
42 };
43 
44 if tx.send(line).await.is_err() {
45 break;
46 }
47 }
48 });
49 
50 let stream: ReceiverStream<Result<String, std::io::Error>> = ReceiverStream::new(rx);
51 let body = Body::from_stream(stream as _);
52 
53 (
54 StatusCode::OK,
55 [(header::CONTENT_TYPE, "application/x-ndjson")],
56 body,
57 )
58 .into_response()
59}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A channel plus a spawned task lets you produce a response body incrementally instead of materializing it all in memory.
  2. 2sqlx's fetch yields rows lazily, so pairing it with a stream keeps memory flat regardless of table size.
  3. 3Bounded channels give you backpressure for free, and a failed send signals the client hung up so the task can stop.

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
go 
package streaming
 
import (
	"bufio"

Streaming NDJSON logs over HTTP in Go

http-streaming channels select
Advanced 10 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

Share this explainer

Here's the card — post it anywhere.

Streaming NDJSON from Postgres in Axum — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code