asynchronous - 使用Rust actix-web的流式传输结果，生命周期问题

Question

我在使用波纹管代码在actix-web上使用Stream时遇到问题:

fn format_csv_row(row: tiberius::Row) -> Result<web::Bytes, ServerError> { ... }

#[get("/stream/")]
async fn get_stream(
    db_pool: web::Data<bb8::Pool<TiberiusConnectionManager>>,
) -> Result<HttpResponse, ServerError> {
    // Get connection
    let mut conn = db_pool
        .get()
        .await
        .map_err(|e| ServerError::PoolUnavailable)?;

    // Execute query
    let stream = conn
        .query("SELECT * FROM table", &[])
        .await
        .map_err(|e| ServerError::QueryFail)?;

    // Build a stream from SQL results
    let stream = stream.map(|x| format_csv_row(x.unwrap()));
    Ok(HttpResponse::Ok().streaming(stream))
}

这是我在此函数中调用的方法:

bb8::Pool::get

tiberius::Client::query

actix-web::HttpResponseBuilder::stream

编译器提示以下错误:

error[E0597]: `db_pool` does not live long enough
   --> src/routes/trades_stream.rs:88:20
    |
88  |       let mut conn = db_pool
    |                      -^^^^^^
    |                      |
    |  ____________________borrowed value does not live long enough
    | |
89  | |         .get()
    | |______________- argument requires that `db_pool` is borrowed for `'static`
...
102 |   }
    |   - `db_pool` dropped here while still borrowed

error[E0597]: `conn` does not live long enough
   --> src/routes/trades_stream.rs:94:18
    |
94  |       let stream = conn
    |                    -^^^
    |                    |
    |  __________________borrowed value does not live long enough
    | |
95  | |         .query("SELECT * FROM table", &[])
    | |__________________________________________- argument requires that `conn` is borrowed for `'static`
...
102 |   }
    |   - `conn` dropped here while still borrowed

error: aborting due to 2 previous errors

我了解到，在流式传输仍在进行时，连接和池已删除。
如何修改我的代码以使其正常工作？
是否可以为 db_pool和 conn添加显式生存期，以使其与 stream相匹配？

Answer 1

要使此工作正常进行，我们需要更改代码，以使流具有对其正在读取的连接的所有权，并且由于bb8的写入方式，您还需要池句柄的所有权。最好的方法是使用 async-stream 条板箱。
我相信这样的事情应该做到:

use async_stream::try_stream;

fn format_csv_row(row: tiberius::Row) -> Result<web::Bytes, ServerError> { ... }

#[get("/stream/")]
async fn get_stream(
    db_pool: web::Data<bb8::Pool<TiberiusConnectionManager>>,
) -> Result<HttpResponse, ServerError> {
    // Cloning a bb8 pool gives a new handle to the same pool.
    let db_pool = db_pool.clone();

    let stream = try_stream! {
        // Get connection
        let mut conn = db_pool
            .get()
            .await
            .map_err(|e| ServerError::PoolUnavailable)?;

        // Execute query
        let stream = conn
            .query("SELECT * FROM table", &[])
            .await
            .map_err(|e| ServerError::QueryFail)?;

        while let Some(row) = stream.next().await {
            yield format_csv_row(row?)?;
        }
    };

    Ok(HttpResponse::Ok().streaming(Box::pin(stream)))
}

您可能需要在 map_err部分上使用另一个 row?。

Is it possible to add explicit lifetime for db_pool and conn to make them match stream ?

不，您不会通过设置生命周期来更改事物的生命周期。取而代之的是，通过更改代码的结构以使其生命周期足够长，来更改生存期。