随着对Rust的 diesel_async的最新改动，它现在支持事务的异步闭包，我想让我的包装器也实现同样的能力，而不是传入一个盒装的Future。

我想把 txw 更新为像下面这样调用它

async fn txw_custom(db: DbContext) -> DbResult<()> {
    db.txw(async |conn| { Ok(conn.batch_execute("SELECT 1").await?) }).await
}

下面是我当前的代码，使用盒装的Future，当前可以工作，但在调用时需要提供一个Box。

use diesel_async::{
    AsyncConnection, AsyncMigrationHarness, AsyncPgConnection,
    pooled_connection::{
        AsyncDieselConnectionManager, ManagerConfig, PoolableConnection, RecyclingMethod,
        bb8::{Pool, PooledConnection, RunError},
    },
};
use futures_util::{FutureExt, future::BoxFuture, try_join};


#[derive(Debug, thiserror::Error)]
pub enum DbError {
    #[error("failed to run migrations because {0}")]
    MigrationError(String),

    #[error("something went wrong when getting connection from the pool because {0}")]
    PoolConnection(#[from] bb8::RunError),

    #[error("failed to execute query because {0}")]
    FailedQuery(#[from] diesel::result::Error),

    #[error("failed to {0}")]
    Other(String),
}

pub type DbResult<T> = std::result::Result<T, DbError>;


#[derive(Clone)]
pub struct DbContext {
    primary_pool: Pool<AsyncPgConnection>,
    replica_pool: Pool<AsyncPgConnection>,
}

impl DbContext {
    pub async fn primary(&self) -> DbResult<PooledConnection<'_, AsyncPgConnection>> {
        self.primary_pool
            .get()
            .await
            .map_err(DbError::PoolConnection)
    }

    pub async fn replica(&self) -> DbResult<PooledConnection<'_, AsyncPgConnection>> {
        self.replica_pool
            .get()
            .await
            .map_err(DbError::PoolConnection)
    }

    /// Run a transaction on a primary connection
    ///
    /// How to use this function:
    /// ```rust
    /// use db::{DbContext, DbResult};
    /// use diesel_async::SimpleAsyncConnection;
    /// async fn txw_custom(db: DbContext) -> DbResult<()> {
    ///     db.txw(|conn| Box::pin(async move { Ok(conn.batch_execute("SELECT 1").await?) })).await
    /// }
    /// ```
    pub async fn txw<F, T>(&self, f: F) -> DbResult<T>
    where
        F: FnOnce(&mut AsyncPgConnection) -> BoxFuture<DbResult<T>> + Send,
        T: Send,
    {
        self.primary()
            .await?
            .transaction(async |c| f(c).await)
            .await
    }

}

解决方案

似乎在这个 GitHub讨论 的帮助下，我找到了答案。

我必须重新添加与diesel_async事务函数相同的trait bound。

pub trait AsyncFunc<T, R>:
    AsyncFnOnce(T) -> R + FnOnce(T) -> <Self as AsyncFunc<T, R>>::Fut
{
    type Fut: Future<Output = R>;
}

impl<F, T, Fut, R> AsyncFunc<T, R> for F
where
    F: AsyncFnOnce(T) -> R + FnOnce(T) -> Fut,
    Fut: Future<Output = R>,
{
    type Fut = Fut;
}

impl DbContext {
    /// Run a transaction on a primary connection
    ///
    /// How to use this function:
    /// ```rust
    /// use db::{DbContext, DbResult};
    /// use diesel_async::SimpleAsyncConnection;
    /// async fn txw_custom(db: DbContext) -> DbResult<()> {
    ///     db.txw(async |conn| Ok(conn.batch_execute("SELECT 1").await?)).await
    /// }
    /// ```
    pub async fn txw<F, T>(&self, f: F) -> DbResult<T>
    where
        F: AsyncFnOnce(&mut AsyncPgConnection) -> DbResult<T>
            + for<'a> AsyncFunc<&'a mut AsyncPgConnection, DbResult<T>, Fut: Send>
            + Send,
        T: Send,
    {
        self.primary()
            .await?
            .transaction(async |c| f(c).await)
            .await
    }
}

我也成功添加了我的自定义事务包装器。

```rust pub struct DbTx<'a> { conn: &'a mut AsyncPgConnection, }

impl<'a> DbTx<'a> { pub(crate) fn new(conn: &'a mut AsyncPgConnection) -> Self { Self { conn } } pub(crate) fn conn(&mut self) -> &mut AsyncPgConnection { self.conn } }

impl DbContext { /// Run a transaction on the primary connection /// /// ```rust /// use db::{DbContext, DbResult}; /// pub async fn using_ptx(db: &DbContext) -> DbResult<()> { /// db.ptx(async |conn| { /// // do something with the connection /// Ok(()) /// }) /// .await /// } pub async fn ptx(&self, f: F) -> DbResult where for<'tx, 'db> F: AsyncFnOnce(&'tx mut DbTx<'db>) -> DbResult + AsyncFunc<&'tx mut DbTx<'db>, DbResult, Fut: Send> + Send, T: Send, { self.primary() .await? .transaction(async |c| f(&mut DbTx::new(c)).await) .await } }