Browse Source

Fix @Transactional support on functions returning Flow

Closes gh-26052
pull/26058/head
Sébastien Deleuze 5 years ago
parent
commit
737d77a739
  1. 6
      spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java
  2. 46
      spring-tx/src/test/kotlin/org/springframework/transaction/annotation/CoroutinesAnnotationTransactionInterceptorTests.kt

6
spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java

@ -352,9 +352,9 @@ public abstract class TransactionAspectSupport implements BeanFactoryAware, Init
} }
return new ReactiveTransactionSupport(adapter); return new ReactiveTransactionSupport(adapter);
}); });
Publisher<?> publisher = (Publisher<?>) txSupport.invokeWithinTransaction(method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm); Object result = txSupport.invokeWithinTransaction(method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm);
return (isSuspendingFunction ? (hasSuspendingFlowReturnType ? KotlinDelegate.asFlow(publisher) : return (isSuspendingFunction ? (hasSuspendingFlowReturnType ? KotlinDelegate.asFlow((Publisher<?>) result) :
KotlinDelegate.awaitSingleOrNull(publisher, ((CoroutinesInvocationCallback) invocation).getContinuation())) : publisher); KotlinDelegate.awaitSingleOrNull((Publisher<?>) result, ((CoroutinesInvocationCallback) invocation).getContinuation())) : result);
} }
PlatformTransactionManager ptm = asPlatformTransactionManager(tm); PlatformTransactionManager ptm = asPlatformTransactionManager(tm);

46
spring-tx/src/test/kotlin/org/springframework/transaction/annotation/CoroutinesAnnotationTransactionInterceptorTests.kt

@ -17,6 +17,9 @@
package org.springframework.transaction.annotation package org.springframework.transaction.annotation
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import org.assertj.core.api.Assertions import org.assertj.core.api.Assertions
import org.junit.jupiter.api.Test import org.junit.jupiter.api.Test
@ -83,14 +86,38 @@ class CoroutinesAnnotationTransactionInterceptorTests {
runBlocking { runBlocking {
try { try {
proxy.suspendingValueFailure() proxy.suspendingValueFailure()
Assertions.fail("No exception thrown as expected")
} }
catch (ex: IllegalStateException) { catch (ex: IllegalStateException) {
} }
} }
assertReactiveGetTransactionAndRollbackCount(1) assertReactiveGetTransactionAndRollbackCount(1)
} }
@Test
fun suspendingFlowSuccess() {
val proxyFactory = ProxyFactory()
proxyFactory.setTarget(TestWithCoroutines())
proxyFactory.addAdvice(TransactionInterceptor(rtm, source))
val proxy = proxyFactory.proxy as TestWithCoroutines
runBlocking {
Assertions.assertThat(proxy.suspendingFlowSuccess().toList()).containsExactly("foo", "foo")
}
assertReactiveGetTransactionAndCommitCount(1)
}
@Test
fun flowSuccess() {
val proxyFactory = ProxyFactory()
proxyFactory.setTarget(TestWithCoroutines())
proxyFactory.addAdvice(TransactionInterceptor(rtm, source))
val proxy = proxyFactory.proxy as TestWithCoroutines
runBlocking {
Assertions.assertThat(proxy.flowSuccess().toList()).containsExactly("foo", "foo")
}
assertReactiveGetTransactionAndCommitCount(1)
}
private fun assertReactiveGetTransactionAndCommitCount(expectedCount: Int) { private fun assertReactiveGetTransactionAndCommitCount(expectedCount: Int) {
Assertions.assertThat(rtm.begun).isEqualTo(expectedCount) Assertions.assertThat(rtm.begun).isEqualTo(expectedCount)
Assertions.assertThat(rtm.commits).isEqualTo(expectedCount) Assertions.assertThat(rtm.commits).isEqualTo(expectedCount)
@ -122,5 +149,22 @@ class CoroutinesAnnotationTransactionInterceptorTests {
delay(10) delay(10)
throw IllegalStateException() throw IllegalStateException()
} }
open fun flowSuccess(): Flow<String> {
return flow {
emit("foo")
delay(10)
emit("foo")
}
}
open suspend fun suspendingFlowSuccess(): Flow<String> {
delay(10)
return flow {
emit("foo")
delay(10)
emit("foo")
}
}
} }
} }

Loading…
Cancel
Save