Skip to content

Commit fc64806

Browse files
committed
Improve Coroutines transaction API
As a follow-up of gh-22915, the purpose of this commit is to improve Coroutines programmatic transaction API to make it more consistent with the Java one and more idiomatic. For suspending functions, this commit changes the TransactionalOperator.transactional extension with a suspending lambda parameter to a TransactionalOperator.executeAndAwait one which is conceptually closer to TransactionalOperator.execute Java API so more consistent. For Flow, the TransactionalOperator.transactional extension is correct but would be more idiomatic as a Flow extension. This commit also adds code samples to the reference documentation. Closes gh-23627
1 parent e62cb6b commit fc64806

File tree

3 files changed

+56
-12
lines changed

3 files changed

+56
-12
lines changed

spring-tx/src/main/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensions.kt

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,23 +7,24 @@ import kotlinx.coroutines.reactive.asFlow
77
import kotlinx.coroutines.reactive.awaitFirstOrNull
88
import kotlinx.coroutines.reactor.asFlux
99
import kotlinx.coroutines.reactor.mono
10+
import org.springframework.transaction.ReactiveTransaction
1011

1112
/**
12-
* Coroutines variant of [TransactionalOperator.transactional] with a [Flow] parameter.
13+
* Coroutines variant of [TransactionalOperator.transactional] as a [Flow] extension.
1314
*
1415
* @author Sebastien Deleuze
1516
* @since 5.2
1617
*/
1718
@ExperimentalCoroutinesApi
18-
fun <T : Any> TransactionalOperator.transactional(flow: Flow<T>): Flow<T> =
19-
transactional(flow.asFlux()).asFlow()
19+
fun <T : Any> Flow<T>.transactional(operator: TransactionalOperator): Flow<T> =
20+
operator.transactional(asFlux()).asFlow()
2021

2122
/**
22-
* Coroutines variant of [TransactionalOperator.transactional] with a suspending lambda
23+
* Coroutines variant of [TransactionalOperator.execute] with a suspending lambda
2324
* parameter.
2425
*
2526
* @author Sebastien Deleuze
2627
* @since 5.2
2728
*/
28-
suspend fun <T : Any> TransactionalOperator.transactional(f: suspend () -> T?): T? =
29-
transactional(mono(Dispatchers.Unconfined) { f() }).awaitFirstOrNull()
29+
suspend fun <T : Any> TransactionalOperator.executeAndAwait(f: suspend (ReactiveTransaction) -> T?): T? =
30+
execute { status -> mono(Dispatchers.Unconfined) { f(status) } }.awaitFirstOrNull()

spring-tx/src/test/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensionsTests.kt

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ class TransactionalOperatorExtensionsTests {
3434
fun commitWithSuspendingFunction() {
3535
val operator = TransactionalOperator.create(tm, DefaultTransactionDefinition())
3636
runBlocking {
37-
operator.transactional {
37+
operator.executeAndAwait {
3838
delay(1)
3939
true
4040
}
@@ -48,7 +48,7 @@ class TransactionalOperatorExtensionsTests {
4848
val operator = TransactionalOperator.create(tm, DefaultTransactionDefinition())
4949
runBlocking {
5050
try {
51-
operator.transactional {
51+
operator.executeAndAwait {
5252
delay(1)
5353
throw IllegalStateException()
5454
}
@@ -72,7 +72,7 @@ class TransactionalOperatorExtensionsTests {
7272
emit(4)
7373
}
7474
runBlocking {
75-
val list = operator.transactional(flow).toList()
75+
val list = flow.transactional(operator).toList()
7676
assertThat(list).hasSize(4)
7777
}
7878
assertThat(tm.commit).isTrue()
@@ -89,7 +89,7 @@ class TransactionalOperatorExtensionsTests {
8989
}
9090
runBlocking {
9191
try {
92-
operator.transactional(flow).toList()
92+
flow.transactional(operator).toList()
9393
} catch (ex: IllegalStateException) {
9494
assertThat(tm.commit).isFalse()
9595
assertThat(tm.rollback).isTrue()

src/docs/asciidoc/languages/kotlin.adoc

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -577,8 +577,51 @@ class UserHandler(builder: WebClient.Builder) {
577577
=== Transactions
578578

579579
Transactions on Coroutines are supported via the programmatic variant of the Reactive
580-
transaction management provided as of Spring Framework 5.2. `TransactionalOperator.transactional`
581-
extensions with suspending lambda and Kotlin `Flow` parameter are provided for that purpose.
580+
transaction management provided as of Spring Framework 5.2.
581+
582+
For suspending functions, a `TransactionalOperator.executeAndAwait` extension is provided.
583+
584+
[source,kotlin,indent=0]
585+
----
586+
import org.springframework.transaction.reactive.executeAndAwait
587+
588+
class PersonRepository(private val operator: TransactionalOperator) {
589+
590+
suspend fun initDatabase() = operator.executeAndAwait {
591+
insertPerson1()
592+
insertPerson2()
593+
}
594+
595+
private suspend fun insertPerson1() {
596+
// INSERT SQL statement
597+
}
598+
599+
private suspend fun insertPerson2() {
600+
// INSERT SQL statement
601+
}
602+
}
603+
----
604+
605+
For Kotlin `Flow`, a `Flow<T>.transactional` extension is provided.
606+
607+
[source,kotlin,indent=0]
608+
----
609+
import org.springframework.transaction.reactive.transactional
610+
611+
class PersonRepository(private val operator: TransactionalOperator) {
612+
613+
fun updatePeople() = findPeople().map(::updatePerson).transactional(operator)
614+
615+
private fun findPeople(): Flow<Person> {
616+
// SELECT SQL statement
617+
}
618+
619+
private suspend fun updatePerson(person: Person): Person {
620+
// UPDATE SQL statement
621+
}
622+
}
623+
----
624+
582625

583626
[[kotlin-spring-projects-in-kotlin]]
584627
== Spring Projects in Kotlin

0 commit comments

Comments
 (0)