使用 r2dbc 驱动程序的 Jooq 反应式抓取

Question

我的查询看起来像（使用 JOOQ v 3.15）：

    override fun getCredentialsById(id: Long): Mono<UserCredentialsModel> {
    return Mono.from {
        dsl.select(
                USER_CREDENTIALS.ID,
                USER_CREDENTIALS.EMAIL,
                USER_CREDENTIALS.PHONE,
                USER_CREDENTIALS.LOGIN,
                USER_CREDENTIALS.PASSWORD)
            .from(USER_CREDENTIALS)
            .where(USER_CREDENTIALS.ID.eq(id))
            .and(USER_CREDENTIALS.IS_ACTIVE.eq(true))
            .fetchInto(UserCredentialsModel::class.java)
    }
}

JOOQ 配置：

@Bean
fun createContext(): DSLContext {
    return DSL.using(connection)
}

其中 connection 是 io.r2dbc.spi.ConnectionFactory

我遇到了一个例外：

org.jooq.exception.DetachedException: Attempt to execute a blocking method (e.g. Query.execute() or ResultQuery.fetch()) when only an R2BDC ConnectionFactory was configured

然后我应该如何以反应式将查询结果提取到 Kotlin 数据 class（或 Java JOOQ 的 POJO）中？

Answer 1

未经测试我会说它应该是

return Mono.from(
    dsl.select(
            USER_CREDENTIALS.ID,
            USER_CREDENTIALS.EMAIL,
            USER_CREDENTIALS.PHONE,
            USER_CREDENTIALS.LOGIN,
            USER_CREDENTIALS.PASSWORD)
        .from(USER_CREDENTIALS)
        .where(USER_CREDENTIALS.ID.eq(id))
        .and(USER_CREDENTIALS.IS_ACTIVE.eq(true)));

Answer 2

当使用 jOOQ 时，你不应该调用任何阻塞方法，例如 ResultQuery.fetchOneInto(Class). That method is just a convenience method for ResultQuery.fetchOne() and then Record.into(Class)。没有什么能阻止您自己调用 Record.into(Class)，无论是在阻塞世界还是非阻塞世界。

因此，使用您常用的反应器库方法来映射流内容：

mono.map { r -> r.into(UserCredentialsModel::class.java) }

或者，在一个完整的例子中：

return Mono.from {
    dsl.select(
            USER_CREDENTIALS.ID,
            USER_CREDENTIALS.EMAIL,
            USER_CREDENTIALS.PHONE,
            USER_CREDENTIALS.LOGIN,
            USER_CREDENTIALS.PASSWORD)
        .from(USER_CREDENTIALS)
        .where(USER_CREDENTIALS.ID.eq(id))
        .and(USER_CREDENTIALS.IS_ACTIVE.eq(true))
} .map { r -> r.into(UserCredentialsModel::class.java) }

Answer 3

你可以试试.toMono().subscribe()

dsl.select(
            USER_CREDENTIALS.ID,
            USER_CREDENTIALS.EMAIL,
            USER_CREDENTIALS.PHONE,
            USER_CREDENTIALS.LOGIN,
            USER_CREDENTIALS.PASSWORD)
        .from(USER_CREDENTIALS)
        .where(USER_CREDENTIALS.ID.eq(id))
        .and(USER_CREDENTIALS.IS_ACTIVE.eq(true))
        .fetchInto(UserCredentialsModel::class.java)
        .toMono().subscribe()

使用 r2dbc 驱动程序的 Jooq 反应式抓取

Jooq reactive fetching using r2dbc driver

jooq

r2dbc