From 1d80cbea35e564a639755a994748cf15dbb9ad2b Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Tue, 7 May 2019 11:08:01 +0200 Subject: [PATCH] Replace signal materialization in TransactionAspectSupport with usingWhen We now use Flux.usingWhen() instead materialize/dematerialize operators to reuse Reactor's resource closure. Until usingWhen() accepts a BiFunction to consume error signals, we need to map error signals outside of usingWhen which requires re-wrapping of the ReactiveTransaction object. Also, reuse the current TransactionContext to leave Transaction creation/propagation entirely to ReactiveTransactionManager instead of creating new TransactionContexts. --- .../interceptor/TransactionAspectSupport.java | 43 +++++++++++-------- .../reactive/TransactionContextManager.java | 2 +- ...bstractReactiveTransactionAspectTests.java | 7 ++- 3 files changed, 32 insertions(+), 20 deletions(-) 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 db051b1089..2175a92da1 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 @@ -834,15 +834,18 @@ public abstract class TransactionAspectSupport implements BeanFactoryAware, Init try { // This is an around advice: Invoke the next interceptor in the chain. // This will normally result in a target object being invoked. - Mono retVal = (Mono) invocation.proceedWithInvocation(); - return retVal - .onErrorResume(ex -> completeTransactionAfterThrowing(it, ex).then(Mono.error(ex))).materialize() - .flatMap(signal -> { - if (signal.isOnComplete() || signal.isOnNext()) { - return commitTransactionAfterReturning(it).thenReturn(signal); - } - return Mono.just(signal); - }).dematerialize(); + // Need re-wrapping of ReactiveTransaction until we get hold of the exception + // through usingWhen. + return Mono.usingWhen(Mono.just(it), s -> { + try { + return (Mono) invocation.proceedWithInvocation(); + } + catch (Throwable throwable) { + return Mono.error(throwable); + } + }, this::commitTransactionAfterReturning, s -> Mono.empty()) + .onErrorResume(ex -> completeTransactionAfterThrowing(it, ex) + .then(Mono.error(ex))); } catch (Throwable ex) { // target invocation exception @@ -860,15 +863,19 @@ public abstract class TransactionAspectSupport implements BeanFactoryAware, Init try { // This is an around advice: Invoke the next interceptor in the chain. // This will normally result in a target object being invoked. - Flux retVal = Flux.from(this.adapter.toPublisher(invocation.proceedWithInvocation())); - return retVal - .onErrorResume(ex -> completeTransactionAfterThrowing(it, ex).then(Mono.error(ex))) - .materialize().flatMap(signal -> { - if (signal.isOnComplete()) { - return commitTransactionAfterReturning(it).materialize(); - } - return Mono.just(signal); - }).dematerialize(); + // Need re-wrapping of ReactiveTransaction until we get hold of the exception + // through usingWhen. + return Flux.usingWhen(Mono.just(it), s -> { + try { + return this.adapter.toPublisher( + invocation.proceedWithInvocation()); + } + catch (Throwable throwable) { + return Mono.error(throwable); + } + }, this::commitTransactionAfterReturning, s -> Mono.empty()) + .onErrorResume(ex -> completeTransactionAfterThrowing(it, ex) + .then(Mono.error(ex))); } catch (Throwable ex) { // target invocation exception diff --git a/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionContextManager.java b/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionContextManager.java index 53dd59f23d..7dc42c5c15 100644 --- a/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionContextManager.java +++ b/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionContextManager.java @@ -88,7 +88,7 @@ public abstract class TransactionContextManager { return context -> { TransactionContextHolder holder = context.get(TransactionContextHolder.class); if (holder.hasContext()) { - context.put(TransactionContext.class, holder.currentContext()); + return context.put(TransactionContext.class, holder.currentContext()); } return context.put(TransactionContext.class, holder.createContext()); }; diff --git a/spring-tx/src/test/java/org/springframework/transaction/interceptor/AbstractReactiveTransactionAspectTests.java b/spring-tx/src/test/java/org/springframework/transaction/interceptor/AbstractReactiveTransactionAspectTests.java index 4e7213c84b..507c9d4e94 100644 --- a/spring-tx/src/test/java/org/springframework/transaction/interceptor/AbstractReactiveTransactionAspectTests.java +++ b/spring-tx/src/test/java/org/springframework/transaction/interceptor/AbstractReactiveTransactionAspectTests.java @@ -33,6 +33,7 @@ import org.springframework.transaction.reactive.TransactionContext; import static org.assertj.core.api.Assertions.*; import static org.assertj.core.api.Fail.fail; +import static org.junit.Assert.*; import static org.mockito.Mockito.*; /** @@ -321,6 +322,7 @@ public abstract class AbstractReactiveTransactionAspectTests { when(rtm.getReactiveTransaction(txatt)).thenReturn(Mono.just(status)); UnexpectedRollbackException ex = new UnexpectedRollbackException("foobar", null); when(rtm.commit(status)).thenReturn(Mono.error(ex)); + when(rtm.rollback(status)).thenReturn(Mono.empty()); DefaultTestBean tb = new DefaultTestBean(); TestBean itb = (TestBean) advised(tb, rtm, tas); @@ -329,7 +331,10 @@ public abstract class AbstractReactiveTransactionAspectTests { Mono.from(itb.setName(name)) .as(StepVerifier::create) - .expectError(UnexpectedRollbackException.class) + .consumeErrorWith(throwable -> { + assertEquals(RuntimeException.class, throwable.getClass()); + assertEquals(ex, throwable.getCause()); + }) .verify(); // Should have invoked target and changed name