如何在 Rust 中使用带有“juniper”订阅的“sqlx”？

Question

背景：

sqlx我在与订阅集成时遇到问题juniper。

我收到了Pin<Box<dyn Stream<Item = Result<User, sqlx::Error>> + 'e + Send>>来自sqlx::query::QueryAs::fetch().

juniper需要将订阅返回为Pin<Box<dyn Stream<Item = Result<User, juniper::FieldError>> + Send>>.

Result<User, sqlx::Error>请注意从到 的更改Result<User, juniper::FieldError>。使用map_err()fromfutures::TryStreamExt，我创建了以下代码来执行查询并转换错误类型。

type UsersStream =
    Pin<Box<dyn Stream<Item = Result<User, FieldError>> + Send>>;

#[juniper::graphql_subscription(Context = Context)]
impl SubscriptionRoot {
    async fn users(context: &Context) -> UsersStream {
        let sqlx::query_as!(User, "SELECT * FROM users")
            .fetch(&context.pool)
            .map_err(|e| {
                FieldError::new(
                    "Database error",
                    graphql_value!(format!("{}", e)))
            })
            .boxed()
    }
}

编译失败并出现以下错误：

error[E0759]: `executor` has lifetime `'ref_e` but it needs to satisfy a `'static` lifetime requirement
  --> server/src/graphql/subscription.rs:27:1
   |
27 |   #[juniper::graphql_subscription(Context = Context)]
   |   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   |   |
   |   this data with lifetime `'ref_e`...
   |   ...is captured here...
...
63 | /         sqlx::query_as!(User, "SELECT * FROM users")
64 | |             .fetch(&context.pool)
65 | |             .map_err(|e| {
66 | |                 FieldError::new(
...  |
69 | |             })
70 | |             .boxed()
   | |____________________- ...and is required to live as long as `'static` here
   |
   = note: this error originates in an attribute macro (in Nightly builds, run with -Z macro-backtrace for more info)

error: aborting due to previous error

我对 s 或生命周期不够熟悉，Stream无法理解此错误的含义。经过更多研究后，似乎这ref_e就是订阅对junipers的引用的生命周期Executor。

尝试：

• 提供生命周期，如graphql-rust/juniper#143juniper::Context中讨论的那样。
• 较高排名的特质界限

版本：

• sqlx-0.4.1
• juniper固定cd66bdb承诺master

Answer 1

您的代码与我的代码不完全一样，但我认为该解决方案也适用于此处，请在使用它之前尝试克隆池：

type UsersStream =
    Pin<Box<dyn Stream<Item = Result<User, FieldError>> + Send>>;

#[juniper::graphql_subscription(Context = Context)]
impl SubscriptionRoot {
    async fn users(context: &Context) -> UsersStream {
        let pool = context.pool.clone();
        let sqlx::query_as!(User, "SELECT * FROM users")
            .fetch(&pool)
            .map_err(|e| {
                FieldError::new(
                    "Database error",
                    graphql_value!(format!("{}", e)))
            })
            .boxed()
    }
}