Axum Streaming Response Body

use axum::{
    body::Body,
    extract::State,
    http::{header, StatusCode},
    response::{IntoResponse, Response},
    routing::get,
    Json, Router,
};
use futures_util::stream::{self, StreamExt};
use std::convert::Infallible;
use std::time::Duration;

async fn export_stream() -> impl futures_util::Stream<Item = Result<String, Infallible>> {
    stream::iter(0..5)
        .then(|i| async move {
            tokio::time::sleep(Duration::from_millis(30)).await;
            Ok(format!("row-{i}\n"))
        })
        .chain(stream::iter(vec![Ok("done\n".to_string())]))
}

async fn export_csv() -> Response {
    let body_stream = export_stream().map(|chunk| {
        chunk
            .map(|text| text.into_bytes())
            .map_err(|err: Infallible| match err {})
    });

    let body = Body::from_stream(body_stream);

    (
        StatusCode::OK,
        [
            (header::CONTENT_TYPE, "text/csv; charset=utf-8"),
            (header::CONTENT_DISPOSITION, "attachment; filename=rows.csv"),
        ],
        body,
    )
        .into_response()
}

async fn json_events() -> Json<Vec<String>> {
    let events: Vec<String> = export_stream()
        .filter_map(|chunk| async move { chunk.ok() })
        .collect()
        .await;
    Json(events)
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/export", get(export_csv))
        .route("/events", get(json_events));

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}