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 std::net::{IpAddr, SocketAddr};
 
use axum::extract::{ConnectInfo, FromRequestParts};
use axum::http::request::Parts;

How a custom ClientIp extractor works in Axum

extractor http-headers proxy-forwarding
Intermediate 8 steps
rust 
use once_cell::sync::Lazy;
use regex::Regex;
 
static LOG_LINE: Lazy<Regex> = Lazy::new(|| {

Parsing access logs with a lazy regex in Rust

regex lazy-initialization parsing
Intermediate 7 steps
java 
public class RequestCoalescer<K, V> {
 
    private final ConcurrentHashMap<K, CompletableFuture<V>> inFlight = new ConcurrentHashMap<>();
    private final Function<K, V> loader;

Coalescing duplicate requests in Java

concurrency caching completablefuture
Advanced 6 steps
rust 
use axum::{
    extract::State,
    response::sse::{Event, KeepAlive, Sse},
    Json,

Proxying an SSE chat stream in Axum

server-sent-events streaming async-generators
Advanced 10 steps
ruby 
require "csv"
 
class CsvExporter
  def initialize(records, columns: nil)

Turning records into CSV in Ruby

csv data-export serialization
Intermediate 7 steps
rust 
use std::collections::HashMap;
 
fn decode_component(input: &str) -> String {
    let bytes = input.as_bytes();

Parsing a URL query string in Rust

url-encoding byte-parsing iterators
Intermediate 8 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