diff --git a/spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java b/spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java index 4c9a27ff15..6752239a6e 100644 --- a/spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java +++ b/spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java @@ -878,7 +878,7 @@ public abstract class TransactionAspectSupport implements BeanFactoryAware, Init }, this::commitTransactionAfterReturning, (txInfo, err) -> Mono.empty(), - this::commitTransactionAfterReturning) + this::rollbackTransactionOnCancel) .onErrorResume(ex -> completeTransactionAfterThrowing(it, ex).then(Mono.error(ex))); } @@ -908,7 +908,7 @@ public abstract class TransactionAspectSupport implements BeanFactoryAware, Init }, this::commitTransactionAfterReturning, (txInfo, ex) -> Mono.empty(), - this::commitTransactionAfterReturning) + this::rollbackTransactionOnCancel) .onErrorResume(ex -> completeTransactionAfterThrowing(it, ex).then(Mono.error(ex))); } @@ -975,6 +975,16 @@ public abstract class TransactionAspectSupport implements BeanFactoryAware, Init return Mono.empty(); } + private Mono rollbackTransactionOnCancel(@Nullable ReactiveTransactionInfo txInfo) { + if (txInfo != null && txInfo.getReactiveTransaction() != null) { + if (logger.isTraceEnabled()) { + logger.trace("Rolling back transaction for [" + txInfo.getJoinpointIdentification() + "] after cancellation"); + } + return txInfo.getTransactionManager().rollback(txInfo.getReactiveTransaction()); + } + return Mono.empty(); + } + private Mono completeTransactionAfterThrowing(@Nullable ReactiveTransactionInfo txInfo, Throwable ex) { if (txInfo != null && txInfo.getReactiveTransaction() != null) { if (logger.isTraceEnabled()) { diff --git a/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java b/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java index 26e343d285..7dc3de8842 100644 --- a/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java +++ b/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java @@ -79,7 +79,7 @@ final class TransactionalOperatorImpl implements TransactionalOperator { // Need re-wrapping of ReactiveTransaction until we get hold of the exception // through usingWhen. return status.flatMap(it -> Mono.usingWhen(Mono.just(it), ignore -> mono, - this.transactionManager::commit, (res, err) -> Mono.empty(), this.transactionManager::commit) + this.transactionManager::commit, (res, err) -> Mono.empty(), this.transactionManager::rollback) .onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex)))); }) .subscriberContext(TransactionContextManager.getOrCreateContext()) @@ -100,7 +100,7 @@ final class TransactionalOperatorImpl implements TransactionalOperator { action::doInTransaction, this.transactionManager::commit, (tx, ex) -> Mono.empty(), - this.transactionManager::commit) + this.transactionManager::rollback) .onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex)))); }) diff --git a/spring-tx/src/main/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensions.kt b/spring-tx/src/main/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensions.kt index 503ee4e2f3..67b93e764c 100644 --- a/spring-tx/src/main/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensions.kt +++ b/spring-tx/src/main/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensions.kt @@ -1,9 +1,10 @@ package org.springframework.transaction.reactive +import java.util.Optional import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.reactive.asFlow -import kotlinx.coroutines.reactive.awaitFirstOrNull +import kotlinx.coroutines.reactive.awaitLast import kotlinx.coroutines.reactor.asFlux import kotlinx.coroutines.reactor.mono import org.springframework.transaction.ReactiveTransaction @@ -22,7 +23,9 @@ fun Flow.transactional(operator: TransactionalOperator): Flow = * parameter. * * @author Sebastien Deleuze +* @author Mark Paluch * @since 5.2 */ suspend fun TransactionalOperator.executeAndAwait(f: suspend (ReactiveTransaction) -> T?): T? = - execute { status -> mono(Dispatchers.Unconfined) { f(status) } }.awaitFirstOrNull() + execute { status -> mono(Dispatchers.Unconfined) { f(status) } }.map { value -> Optional.of(value) } + .defaultIfEmpty(Optional.empty()).awaitLast().orElse(null) diff --git a/spring-tx/src/test/java/org/springframework/transaction/annotation/AnnotationTransactionInterceptorTests.java b/spring-tx/src/test/java/org/springframework/transaction/annotation/AnnotationTransactionInterceptorTests.java index c7477db015..e026e4ad92 100644 --- a/spring-tx/src/test/java/org/springframework/transaction/annotation/AnnotationTransactionInterceptorTests.java +++ b/spring-tx/src/test/java/org/springframework/transaction/annotation/AnnotationTransactionInterceptorTests.java @@ -16,13 +16,19 @@ package org.springframework.transaction.annotation; +import java.time.Duration; + import io.vavr.control.Try; import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import org.springframework.aop.framework.ProxyFactory; import org.springframework.transaction.interceptor.TransactionInterceptor; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.testfixture.CallCountingTransactionManager; +import org.springframework.transaction.testfixture.ReactiveCallCountingTransactionManager; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; @@ -32,11 +38,14 @@ import static org.assertj.core.api.Assertions.assertThatIllegalStateException; /** * @author Rob Harrop * @author Juergen Hoeller + * @author Mark Paluch */ public class AnnotationTransactionInterceptorTests { private final CallCountingTransactionManager ptm = new CallCountingTransactionManager(); + private final ReactiveCallCountingTransactionManager rtm = new ReactiveCallCountingTransactionManager(); + private final AnnotationTransactionAttributeSource source = new AnnotationTransactionAttributeSource(); private final TransactionInterceptor ti = new TransactionInterceptor(this.ptm, this.source); @@ -169,6 +178,78 @@ public class AnnotationTransactionInterceptorTests { .satisfies(ex -> assertGetTransactionAndRollbackCount(1)); } + @Test + public void withMonoSuccess() { + ProxyFactory proxyFactory = new ProxyFactory(); + proxyFactory.setTarget(new TestWithReactive()); + proxyFactory.addAdvice(new TransactionInterceptor(rtm, this.source)); + + TestWithReactive proxy = (TestWithReactive) proxyFactory.getProxy(); + + StepVerifier.withVirtualTime(proxy::monoSuccess).thenAwait(Duration.ofSeconds(10)).verifyComplete(); + assertReactiveGetTransactionAndCommitCount(1); + } + + @Test + public void withMonoFailure() { + ProxyFactory proxyFactory = new ProxyFactory(); + proxyFactory.setTarget(new TestWithReactive()); + proxyFactory.addAdvice(new TransactionInterceptor(rtm, this.source)); + + TestWithReactive proxy = (TestWithReactive) proxyFactory.getProxy(); + + proxy.monoFailure().as(StepVerifier::create).verifyError(); + assertReactiveGetTransactionAndRollbackCount(1); + } + + @Test + public void withMonoRollback() { + ProxyFactory proxyFactory = new ProxyFactory(); + proxyFactory.setTarget(new TestWithReactive()); + proxyFactory.addAdvice(new TransactionInterceptor(rtm, this.source)); + + TestWithReactive proxy = (TestWithReactive) proxyFactory.getProxy(); + + StepVerifier.withVirtualTime(proxy::monoSuccess).thenAwait(Duration.ofSeconds(1)).thenCancel().verify(); + assertReactiveGetTransactionAndRollbackCount(1); + } + + @Test + public void withFluxSuccess() { + ProxyFactory proxyFactory = new ProxyFactory(); + proxyFactory.setTarget(new TestWithReactive()); + proxyFactory.addAdvice(new TransactionInterceptor(rtm, this.source)); + + TestWithReactive proxy = (TestWithReactive) proxyFactory.getProxy(); + + StepVerifier.withVirtualTime(proxy::fluxSuccess).thenAwait(Duration.ofSeconds(10)).expectNextCount(1).verifyComplete(); + assertReactiveGetTransactionAndCommitCount(1); + } + + @Test + public void withFluxFailure() { + ProxyFactory proxyFactory = new ProxyFactory(); + proxyFactory.setTarget(new TestWithReactive()); + proxyFactory.addAdvice(new TransactionInterceptor(rtm, this.source)); + + TestWithReactive proxy = (TestWithReactive) proxyFactory.getProxy(); + + proxy.fluxFailure().as(StepVerifier::create).verifyError(); + assertReactiveGetTransactionAndRollbackCount(1); + } + + @Test + public void withFluxRollback() { + ProxyFactory proxyFactory = new ProxyFactory(); + proxyFactory.setTarget(new TestWithReactive()); + proxyFactory.addAdvice(new TransactionInterceptor(rtm, this.source)); + + TestWithReactive proxy = (TestWithReactive) proxyFactory.getProxy(); + + StepVerifier.withVirtualTime(proxy::fluxSuccess).thenAwait(Duration.ofSeconds(1)).thenCancel().verify(); + assertReactiveGetTransactionAndRollbackCount(1); + } + @Test public void withVavrTrySuccess() { ProxyFactory proxyFactory = new ProxyFactory(); @@ -342,6 +423,16 @@ public class AnnotationTransactionInterceptorTests { assertThat(this.ptm.rollbacks).isEqualTo(expectedCount); } + private void assertReactiveGetTransactionAndCommitCount(int expectedCount) { + assertThat(this.rtm.begun).isEqualTo(expectedCount); + assertThat(this.rtm.commits).isEqualTo(expectedCount); + } + + private void assertReactiveGetTransactionAndRollbackCount(int expectedCount) { + assertThat(this.rtm.begun).isEqualTo(expectedCount); + assertThat(this.rtm.rollbacks).isEqualTo(expectedCount); + } + @Transactional public static class TestClassLevelOnly { @@ -452,6 +543,25 @@ public class AnnotationTransactionInterceptorTests { } } + @Transactional + public static class TestWithReactive { + + public Mono monoSuccess() { + return Mono.delay(Duration.ofSeconds(10)).then(); + } + + public Mono monoFailure() { + return Mono.error(new IllegalStateException()); + } + + public Flux fluxSuccess() { + return Flux.just(new Object()).delayElements(Duration.ofSeconds(10)); + } + + public Flux fluxFailure() { + return Flux.error(new IllegalStateException()); + } + } @Transactional public static class TestWithVavrTry { diff --git a/spring-tx/src/test/java/org/springframework/transaction/reactive/TransactionalOperatorTests.java b/spring-tx/src/test/java/org/springframework/transaction/reactive/TransactionalOperatorTests.java index 8a78557376..a0827bcd79 100644 --- a/spring-tx/src/test/java/org/springframework/transaction/reactive/TransactionalOperatorTests.java +++ b/spring-tx/src/test/java/org/springframework/transaction/reactive/TransactionalOperatorTests.java @@ -70,8 +70,8 @@ public class TransactionalOperatorTests { .thenAwait() .thenCancel() .verify(); - assertThat(tm.commit).isTrue(); - assertThat(tm.rollback).isFalse(); + assertThat(tm.commit).isFalse(); + assertThat(tm.rollback).isTrue(); assertThat(cancelled).isTrue(); } @@ -84,8 +84,8 @@ public class TransactionalOperatorTests { .thenAwait() .thenCancel() .verify(); - assertThat(tm.commit).isTrue(); - assertThat(tm.rollback).isFalse(); + assertThat(tm.commit).isFalse(); + assertThat(tm.rollback).isTrue(); assertThat(cancelled).isTrue(); } diff --git a/spring-tx/src/test/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensionsTests.kt b/spring-tx/src/test/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensionsTests.kt index f42cb94057..96384f939d 100644 --- a/spring-tx/src/test/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensionsTests.kt +++ b/spring-tx/src/test/kotlin/org/springframework/transaction/reactive/TransactionalOperatorExtensionsTests.kt @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -42,6 +42,19 @@ class TransactionalOperatorExtensionsTests { assertThat(tm.rollback).isFalse() } + @Test + fun commitWithEmptySuspendingFunction() { + val operator = TransactionalOperator.create(tm, DefaultTransactionDefinition()) + runBlocking { + operator.executeAndAwait { + delay(1) + null + } + } + assertThat(tm.commit).isTrue() + assertThat(tm.rollback).isFalse() + } + @Test fun rollbackWithSuspendingFunction() { val operator = TransactionalOperator.create(tm, DefaultTransactionDefinition()) diff --git a/spring-tx/src/testFixtures/java/org/springframework/transaction/testfixture/ReactiveCallCountingTransactionManager.java b/spring-tx/src/testFixtures/java/org/springframework/transaction/testfixture/ReactiveCallCountingTransactionManager.java new file mode 100644 index 0000000000..d926458b00 --- /dev/null +++ b/spring-tx/src/testFixtures/java/org/springframework/transaction/testfixture/ReactiveCallCountingTransactionManager.java @@ -0,0 +1,71 @@ +/* + * Copyright 2002-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.transaction.testfixture; + +import reactor.core.publisher.Mono; + +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionException; +import org.springframework.transaction.reactive.AbstractReactiveTransactionManager; +import org.springframework.transaction.reactive.GenericReactiveTransaction; +import org.springframework.transaction.reactive.TransactionSynchronizationManager; + +/** + * @author Mark Paluch + */ +@SuppressWarnings("serial") +public class ReactiveCallCountingTransactionManager extends AbstractReactiveTransactionManager { + + public TransactionDefinition lastDefinition; + public int begun; + public int commits; + public int rollbacks; + public int inflight; + + @Override + protected Object doGetTransaction(TransactionSynchronizationManager synchronizationManager) throws TransactionException { + return new Object(); + } + + @Override + protected Mono doBegin(TransactionSynchronizationManager synchronizationManager, Object transaction, TransactionDefinition definition) throws TransactionException { + this.lastDefinition = definition; + ++begun; + ++inflight; + return Mono.empty(); + } + + @Override + protected Mono doCommit(TransactionSynchronizationManager synchronizationManager, GenericReactiveTransaction status) throws TransactionException { + ++commits; + --inflight; + return Mono.empty(); + } + + @Override + protected Mono doRollback(TransactionSynchronizationManager synchronizationManager, GenericReactiveTransaction status) throws TransactionException { + ++rollbacks; + --inflight; + return Mono.empty(); + } + + + public void clear() { + begun = commits = rollbacks = inflight = 0; + } + +} diff --git a/src/docs/asciidoc/data-access.adoc b/src/docs/asciidoc/data-access.adoc index 4086e4ddd3..f7d0c3c441 100644 --- a/src/docs/asciidoc/data-access.adoc +++ b/src/docs/asciidoc/data-access.adoc @@ -184,7 +184,7 @@ transaction management. The following listing shows the definition of the @Throws(TransactionException::class) fun rollback(status: TransactionStatus) } ----- +---- This is primarily a service provider interface (SPI), although you can use it <> from your application code. Because @@ -241,7 +241,7 @@ listing shows the transaction strategy defined by @Throws(TransactionException::class) fun rollback(status: ReactiveTransaction): Mono } ----- +---- The reactive transaction manager is primarily a service provider interface (SPI), although you can use it <> from your @@ -1698,7 +1698,7 @@ in the application context: @Transactional("account") public void doSomething() { ... } - + @Transactional("reactive-account") public Mono doSomethingReactive() { ... } } @@ -2399,11 +2399,11 @@ the `TransactionOperator` resembles the next example: } public Mono someServiceMethod() { - + // the code in this method executes in a transactional context - + Mono update = updateOperation1(); - + return update.then(resultOfUpdateOperation2).as(transactionalOperator::transactional); } } @@ -2463,9 +2463,7 @@ In Reactive Streams, a `Subscriber` can cancel its `Subscription` and terminate `Publisher`. Operators in Project Reactor, as well as in other libraries, such as `next()`, `take(long)`, `timeout(Duration)`, and others can issue cancellations. There is no way to know the reason for the cancellation, whether it is due to an error or a simply lack of -interest to consume further, and in version 5.2 the `TransactionalOperator` defaults to -committing the transaction on cancel. In version 5.3 this behavior will change and -transactions will be roll back on cancel to create a reliable and deterministic outcome. +interest to consume further. Since version 5.3 cancel signals lead to a roll back. As a result it is important to consider the operators used downstream from a transaction `Publisher`. In particular in the case of a `Flux` or other multi-value `Publisher`, the full output must be consumed to allow the transaction to complete. @@ -2490,7 +2488,7 @@ following example shows customization of the transactional settings for a specif public SimpleService(ReactiveTransactionManager transactionManager) { DefaultTransactionDefinition definition = new DefaultTransactionDefinition(); - + // the transaction settings can be set here explicitly if so desired definition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_UNCOMMITTED); definition.setTimeout(30); // 30 seconds @@ -2588,9 +2586,9 @@ following example shows how to do so: def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED); Mono reactiveTx = txManager.getReactiveTransaction(def); - + reactiveTx.flatMap(status -> { - + Mono tx = ...; // execute your business logic here return tx.then(txManager.commit(status)) @@ -2942,7 +2940,7 @@ this `DataSource`. The following example autowires a `DataSource`: ---- @Repository class JdbcMovieFinder(dataSource: DataSource) : MovieFinder { - + private val jdbcTemplate = JdbcTemplate(dataSource) // ... @@ -3202,8 +3200,8 @@ The following query finds and populates a single domain object: ---- val actor = jdbcTemplate.queryForObject( "select first_name, last_name from t_actor where id = ?", - arrayOf(1212L)) { rs, _ -> - Actor(rs.getString("first_name"), rs.getString("last_name")) + arrayOf(1212L)) { rs, _ -> + Actor(rs.getString("first_name"), rs.getString("last_name")) } ---- @@ -3455,7 +3453,7 @@ method with `@Autowired`. The following example shows how to do so: class JdbcCorporateEventDao(dataSource: DataSource) : CorporateEventDao { // <2> private val jdbcTemplate = JdbcTemplate(dataSource) // <3> - + // JDBC-backed implementations of the methods on the CorporateEventDao follow... } ---- @@ -3794,10 +3792,10 @@ translator: private val jdbcTemplate = JdbcTemplate(dataSource).apply { // create a custom translator and set the DataSource for the default translation lookup exceptionTranslator = CustomSQLErrorCodesTranslator().apply { - this.dataSource = dataSource + this.dataSource = dataSource } } - + fun updateShippingCharge(orderId: Long, pct: Long) { // use the prepared JdbcTemplate for this update this.jdbcTemplate!!.update("update orders" + @@ -4021,8 +4019,8 @@ on Oracle but may not work on other platforms: val name = "Rob" val keyHolder = GeneratedKeyHolder() - jdbcTemplate.update({ - it.prepareStatement (INSERT_SQL, arrayOf("id")).apply { setString(1, name) } + jdbcTemplate.update({ + it.prepareStatement (INSERT_SQL, arrayOf("id")).apply { setString(1, name) } }, keyHolder) // keyHolder.getKey() now contains the generated key @@ -4960,7 +4958,7 @@ the constructor of your `SimpleJdbcCall`. The following example shows this confi private var procReadActor = SimpleJdbcCall(JdbcTemplate(dataSource).apply { isResultsMapCaseInsensitive = true }).withProcedureName("read_actor") - + // ... additional methods } ---- @@ -5718,7 +5716,7 @@ the supplied `ResultSet`, as follows: import org.springframework.jdbc.core.RowMapper class GenreMapper : RowMapper { - + override fun mapRow(rs: ResultSet, rowNum: Int): Genre { return Genre(rs.getString("name")) } @@ -6836,7 +6834,7 @@ implementation resembles the following example, based on the plain Hibernate API .Kotlin ---- class ProductDaoImpl(private val sessionFactory: SessionFactory) : ProductDao { - + fun loadProductsByCategory(category: String): Collection<*> { return sessionFactory.currentSession .createQuery("from test.Product product where product.category=?") @@ -7044,7 +7042,7 @@ and an example for a business method implementation: ---- class ProductServiceImpl(transactionManager: PlatformTransactionManager, private val productDao: ProductDao) : ProductService { - + private val transactionTemplate = TransactionTemplate(transactionManager) fun increasePriceOfAllProductsInCategory(category: String) {