Replace signal materialization in TransactionalOperator 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.

Author: mp911de

spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java

@@ -78,14 +78,11 @@ public <T> Flux<T> execute(TransactionCallback<T> action) throws TransactionExce
 			return status.flatMapMany(it -> {
 				// This is an around advice: Invoke the next interceptor in the chain.
 				// This will normally result in a target object being invoked.
-				Flux<Object> retVal = Flux.from(action.doInTransaction(it));
-				return retVal.onErrorResume(ex -> rollbackOnException(it, ex).
-						then(Mono.error(ex))).materialize().flatMap(signal -> {
-							if (signal.isOnComplete()) {
-								return this.transactionManager.commit(it).materialize();
-							}
-							return Mono.just(signal);
-				}).<T>dematerialize();
+				// Need re-wrapping of ReactiveTransaction until we get hold of the exception
+				// through usingWhen.
+				return Flux.usingWhen(Mono.just(it), action::doInTransaction,
+						this.transactionManager::commit, s -> Mono.empty())
+						.onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex)));
 			});
 		})
 		.subscriberContext(TransactionContextManager.getOrCreateContext())

spring-tx/src/test/java/org/springframework/transaction/reactive/TransactionalOperatorTests.java

@@ -59,9 +59,9 @@ public void rollbackWithMono() {
 	@Test
 	public void commitWithFlux() {
 		TransactionalOperator operator = TransactionalOperator.create(tm, new DefaultTransactionDefinition());
-		Flux.just(true).as(operator::transactional)
+		Flux.just(1, 2, 3, 4).as(operator::transactional)
 				.as(StepVerifier::create)
-				.expectNext(true)
+				.expectNextCount(4)
 				.verifyComplete();
 		assertTrue(tm.commit);
 		assertFalse(tm.rollback);