Combining async transaction and streaming

[AbdulWMughal]

@weiznich How would you construct a handler that uses both diesel-async transaction and streaming together. An example including one of the built in connection poolings would be helpful

Here’s my attempt could you correct it please. I’m after 100% async & non blocking. I’m sure there’s a much more efficient way to use transaction and streaming simultaneously:

async fn complex_query_handler(pool: web::Data<Pool<diesel_async::AsyncPgConnection>>) -> impl Responder {
    let conn = pool.get().await;
    if let Err(e) = conn {
        return HttpResponse::InternalServerError().body(format!("Failed to get DB connection: {:?}", e));
    }
    let mut conn = conn.unwrap();

    let result = conn.transaction::<_, Error, _>(|conn| {
        async move {
            // Insert a new user
            diesel::insert_into(users::table)
                .values((users::name.eq("Ruby"),))
                .execute(conn)
                .await?;

            // Stream all user names starting with 'L' or ID > 0
            let user_stream: impl Stream<Item = Result<User, diesel::result::Error>> = users::table
                .filter(users::id.gt(0))
                .or_filter(users::name.like("%L%"))
                .select(User::as_select())
                .load_stream(&mut conn)
                .await?;

            // Collect the streamed results into a vector
            let users_vec: Vec<User> = user_stream
                .collect::<Result<Vec<_>, _>>()
                .await?;

            Ok(users_vec)
        }
        .scope_boxed()
    }).await;

    match result {
        Ok(users) => HttpResponse::Ok().json(users),
        Err(e) => HttpResponse::InternalServerError().body(format!("Transaction error: {:?}", e)),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    dotenv().ok();
    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL must be set");
    let config = AsyncDieselConnectionManager::<diesel_async::AsyncPgConnection>::new(database_url);
    let pool = Pool::builder(config).build().expect("Failed to create pool.");

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(pool.clone()))
            .route("/complex_query", web::get().to(complex_query_handler))
    })
    .bind("127.0.0.1:8080")?
    .run()
    .await
}

Thanks

[weiznich]

It's generally not meaningful to use load_stream in that way as the ordinary load function does exactly that internally. The load_stream function is designed for usecases where you want to process returned values one by one (or at least by batches). For cases where you just collect everything into a single vector you can just use the existing load function.

Otherwise: Performance is always highly dependent on the context of your application. Therefore it's not really possible to others to answer this kind of generic questions. If you need performance help you need to sit down and describe what specific problem you are trying to solve, what measurements are relevant and how you think they should behave instead.

[AbdulWMughal]

Can you modify the example to use load_stream in the most meaningful way?

kind Regards

[weiznich]

As written before: It's not meaningful to use load_stream in that case at all.

[ag039st]

I've similar issue where I never succeeded to use the async-diesel with actix web if I inject the connection into app Data

[weiznich]

@ag039st If you want to have help with anything I suggest that you open a new discussion thread with your question. Such a post should contain at least which code you've already written and what's the exact problem you have. Nobody can help you solving "it doesn't work" situation.
Ideally you provide a self contained minimal example of your problem.