How to return streams and battle self-referential issues?

[milianw]

Hello all,

could someone be so kind and document the recommended way to integrate sea-orm streams with other libraries? In my case e.g. I would like to integrate with axum / axum_streams.

My current attempt can be found at the links below, and it uses the approach introduced by https://morestina.net/blog/1868/self-referential-types-for-fun-and-profit

The gist is the following:

• async SQL query stream is mapped into a known type (https://github.com/milianw/powerlog/blob/main/src/lib.rs#L388)
• said stream is then turned into a JSON stream via StreamBodyAs from axum_streams (https://github.com/milianw/powerlog/blob/main/src/bin/api.rs#L68)
• the problem now is that I need to return the StreamBodyAs but the SQL query stream references the local DatabaseConnection. I thus need to return a self-referential struct AsyncDbResponse and add impl IntoResponse for AsyncDbResponse (https://github.com/milianw/powerlog/blob/main/src/bin/api.rs#L50)

Is that really the way to go? To a rust novice like me that sounds pretty backwards and smells like I'm either doing something fundamentally wrong, or that the two libraries at play don't work well together. Could someone shed some light on this? What is the recommended way to efficiently turn a stream of SQL queries into a stream of e.g. JSON output?

For completeness sake, I also tried to use orouboros as that is apparently what sea-orm uses internally? But I cannot make heads and tails of how to apply that here, e.g. this is the best I can come up with and it has still lots of problems that don't compile:

#[self_referencing]
struct AsyncDbResponse {
    db: sea_orm::DatabaseConnection,
    #[borrows(mut db)]
    #[not_covariant]
    stream: StreamBodyAs<'this>
}

impl IntoResponse for AsyncDbResponse {
    fn into_response(self) -> Response {
        // FTBFS:
        // cannot move out of `*stream` which is behind a shared reference
        // move occurs because `*stream` has type `axum_streams::StreamBodyAs<'_>`, which does not implement the `Copy` trait
        self.with_stream(|stream| stream.into_response())
    }
}

async fn power_today(State(state): State<Arc<AppState>>) -> Result<impl IntoResponse, AppError> {
    let db = state.db.clone();
    // FTBFS: `db` does not live long enough
    // I guess I need to build this stream in the `stream_builder` lambda below, but I cannot use `.await?` therein...
    let db_stream = db::select_power_today(&db).await?;
    Ok(AsyncDbResponseBuilder{
        db,
        stream_builder: |db| {
            StreamBodyAsOptions::new()
                .buffering_ready_items(1000)
                .json_array(db_stream)
        }
    }.build())
}

Thanks