Lifetime issue with axum-streams - working

axum-streams-lifetime.rs

use axum::extract::State;
use axum::response::IntoResponse;
use axum::{Router, extract::FromRef, routing::get};
use axum_streams::StreamBodyAs;
use futures_core::Stream;
use futures_util::StreamExt;
use serde::Serialize;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{Executor, Row, SqlitePool};
use std::str::FromStr;
use std::sync::Arc;

#[derive(Clone)]
pub struct Config {
    pub external_url: Arc<str>,
}

#[derive(Serialize)]
struct CsvRow {
    id: i64,
    url: String,
}

#[derive(Clone, FromRef)]
pub struct AppState {
    pub db_pool: SqlitePool,
    pub config: Config,
}

/// Stream the result from the database and filter out Error Result (I don't mind for this example)
fn get_db_stream(db_pool: &SqlitePool) -> impl Stream<Item = i64> + 'static {
    let data = db_pool.fetch("SELECT 1 UNION SELECT 2");
    data.filter_map(async |a| a.map(|row| row.get::<'_, i64, usize>(0)).ok())
}
/// Transform the ids into a struct that can be serialized as a csv record
fn get_id_and_url(
    db_pool: &SqlitePool,
    external_url: Arc<str>,
) -> impl Stream<Item = CsvRow> + 'static {
    let data = get_db_stream(db_pool);
    data.map(move |id| {
        let url = format!("{}/users/{}", external_url, id);
        CsvRow { id, url }
    })
}

async fn stream_csv(
    State(db_pool): State<SqlitePool>,
    State(config): State<Config>,
) -> impl IntoResponse {
    let data = get_id_and_url(&db_pool, Arc::clone(&config.external_url));
    StreamBodyAs::csv(data)
}

/// You'd need at least the following deps, I might be missing some:
///
/// tokio = { version = "1.50.0", features = ["full"] }
/// axum = { version ="0.8.8", features = ["macros"] }
/// axum-streams = { version = "0.25.0", features = ["csv"] }
/// futures-core = "0.3.32"
/// futures-util = "0.3.32"
/// serde = { version = "1.0", features = ["derive"] }
#[tokio::main]
async fn main() {
    let options = SqliteConnectOptions::from_str("sqlite://")
        .unwrap()
        .foreign_keys(true)
        .create_if_missing(false)
        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);

    let pool = SqlitePool::connect_with(options).await.unwrap();

    let config = Config {
        external_url: Arc::from("http://localhost:3000"),
    };

    let app_state = AppState {
        db_pool: pool,
        config: config,
    };

    let app = Router::new()
        .route("/test", get(stream_csv))
        .with_state(app_state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
        .await
        .unwrap();
    println!("Listening on: {}", listener.local_addr().unwrap());
    axum::serve(listener, app).await.unwrap();
}

see related discussion: transact-rs/sqlx#3019