Is there a Kotlin-coroutine-friendly way to do transactions with jOOQ?

user_90848 0
Viewed 429

Sometimes we want to have a suspend function that calls other suspend functions inside of a transaction:

suspend fun mySuspendFunction(jooqContext: DSLContext) {
    jooqContext.transaction { config ->
        val transactionContext: DSLContext = config.dsl()

        // ... some code that uses transactionContext ...
        anotherSuspendFunction(transactionContext)
        // ... more code that uses transactionContext ...
    }
}

However, this won't compile, because jOOQ's DSLContext.transaction takes a non-suspending function as its parameter, and so the body of the transaction cannot call a suspend function directly. The above will fail to compile with an error like:

Suspension functions can be called only within coroutine body

One workaround is to use runBlocking around the suspend calls:

suspend fun mySuspendFunction(jooqContext: DSLContext) {
    jooqContext.transaction { config ->
        val transactionContext: DSLContext = config.dsl()

        // ... some code that uses transactionContext ...
        runBlocking {
            anotherSuspendFunction(transactionContext)
        }
        // ... more code that uses transactionContext ...
    }
}

However the documentation for runBlocking advises against using it in this way ("designed to bridge regular blocking code to libraries that are written in suspending style, to be used in main functions and in tests").

That's just a hint of the real problem, which is that because DSLContext.transaction is calling a non-suspend function for the transaction body, the coroutine's thread is blocking (ie: cannot suspend) for the duration of the transaction.

I thought about making a suspend-friendly transaction wrapper for Kotlin, but jOOQ does not yet have a procedural transaction API. Also, DSLContext.connection blocks on the supplied function so I don't see a way to get a hold of the single Connection for the duration of the transaction without blocking.

Is there a way to use a jOOQ transaction in a Kotlin coroutine that does not block for the duration of the transaction, and which allows suspend functions to be called from within the transaction body?

jooq kotlin
2 Answers

jOOQ 3.16

jOOQ 3.16 does not yet have a reactive transaction API. This will be addressed with jOOQ 3.17, soon: https://github.com/jOOQ/jOOQ/issues/13502

jOOQ 3.17 bridging between reactive and coroutines manually

With jOOQ 3.17, you can write:

suspend fun mySuspendFunction(jooqContext: DSLContext): Any {
    return jooqContext.transactionPublisher { config ->

        // Turn the suspension result into a Mono, which implements the reactive
        // streams Publisher<T> SPI, which jOOQ expects as a result from a 
        // TransactionalPublishable
        mono {
            anotherSuspendFunction(config)
        }
    }

    // Turn the Publisher<T> that is returned from transactionPublisher() back 
    // into a suspension result
    .awaitFirst()
}

This is assuming, you're using the recommended reactive streams / coroutines bridges:

<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-coroutines-core</artifactId>
    <version>${kotlinx.coroutines.version}</version>
</dependency>
<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-coroutines-reactor</artifactId>
    <version>${kotlinx.coroutines.version}</version>
</dependency>

jOOQ 3.17, alternatively, using jooq-kotlin-coroutines

With https://github.com/jOOQ/jOOQ/issues/9335, there will be a new module jooq-kotlin-coroutines, which you can pull in to bridge between the worlds for you, e.g.

suspend fun mySuspendFunction(jooqContext: DSLContext) {
    jooqContext.transactionCoroutine { config ->
        anotherSuspendFunction(config)
    }
}

This just removes the glue code for you

jOOQ transaction APIs are just convenience. You don't need them

Notice that you don't need jOOQ's transaction API, nor do you need it to be "procedural." All of this API is just convenience on top of JDBC or R2DBC. You can always access your R2DBC Connection directly, and invoke its transaction API directly to achieve non-blocking transactional behaviour. It's just a bit more glue code.

user_5217990

I will write about the way I know how to do it without using the API provided by jOOQ.

Here is how to make the Transactional annotation work with an R2DBC connection. https://github.com/jOOQ/jOOQ/issues/12218#issuecomment-1213929029

You can also use TransactionalOperator to perform transactions manually if you use jOOQ with a connection that Spring manages transactions in the way described above.

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.mono
import org.springframework.stereotype.Component
import org.springframework.transaction.reactive.TransactionalOperator

@Component
class TransactionOperation(private val transactionalOperator: TransactionalOperator) {
    suspend operator fun <T> invoke(block: suspend CoroutineScope.() -> T): T = transactionalOperator
        .transactional(mono(block = block))
        .awaitSingle()
}

@Component
class Foo(private val transactionOperation: TransactionOperation) {
    suspend fun saveByJooq(entity: Any): Any { TODO("Saving process here.") }

    suspend fun call() {
        // call saving process with transaction.
        transactionOperation { saveByJooq("foo") }
    }
}
user_137097490
Related