Add Reactive TX Manager support (#3201)

* Add Reactive TX Manager support

* Change `TransactionInterceptorBuilder` and `TransactionHandleMessageAdvice` to reply
on a generic `TransactionManager`, so we can configure reactive one as well
* Change a `<transactional>` XML element to support a generic `TransactionManager` reference,
so we can configure reactive one as well
* Support `adviceChain` configuration for the `ReactiveMessageHandler` in the `ConsumerEndpointFactoryBean`
* Attemp to apply reactive transaction for the Reactive MongoDB channel adapter

* * Revert `MongoDbTests` change -
to support transactions we need the latest MongoDb server with replica enabled
* Document reactive transactions

* * Address PR reviews

spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

@@ -200,10 +200,12 @@ public void afterPropertiesSet() {
 		}
 
 		if (!(this.handler instanceof ReactiveMessageHandlerAdapter)) {
-			adviceChain();
+			this.handler = adviceChain(this.handler);
 		}
 		else if (!CollectionUtils.isEmpty(this.adviceChain)) {
-			LOGGER.warn("the advice chain cannot be applied to a 'ReactiveMessageHandler'");
+			this.handler =
+					new ReactiveMessageHandlerAdapter(
+							adviceChain(((ReactiveMessageHandlerAdapter) this.handler).getDelegate()));
 		}
 		if (this.channelResolver == null) {
 			this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory);
@@ -234,7 +236,9 @@ private void populateComponentNameIfAny() {
 		}
 	}
 
-	private void adviceChain() {
+	@SuppressWarnings("unchecked")
+	private <H> H adviceChain(H handler) {
+		H theHandler = handler;
 		if (!CollectionUtils.isEmpty(this.adviceChain)) {
 			/*
 			 *  ARPMHs advise the handleRequestMessage method internally and already have the advice chain injected.
@@ -243,24 +247,25 @@ private void adviceChain() {
 			 *  If the handler is already advised,
 			 *  add the configured advices to its chain, otherwise create a proxy.
 			 */
-			Class<?> targetClass = AopUtils.getTargetClass(this.handler);
+			Class<?> targetClass = AopUtils.getTargetClass(theHandler);
 			boolean replyMessageHandler = AbstractReplyProducingMessageHandler.class.isAssignableFrom(targetClass);
 
 			for (Advice advice : this.adviceChain) {
 				if (!replyMessageHandler || advice instanceof HandleMessageAdvice) {
 					NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice);
 					handlerAdvice.addMethodName("handleMessage");
-					if (this.handler instanceof Advised) {
-						((Advised) this.handler).addAdvisor(handlerAdvice);
+					if (theHandler instanceof Advised) {
+						((Advised) theHandler).addAdvisor(handlerAdvice);
 					}
 					else {
-						ProxyFactory proxyFactory = new ProxyFactory(this.handler);
+						ProxyFactory proxyFactory = new ProxyFactory(theHandler);
 						proxyFactory.addAdvisor(handlerAdvice);
-						this.handler = (MessageHandler) proxyFactory.getProxy(this.beanClassLoader);
+						theHandler = (H) proxyFactory.getProxy(this.beanClassLoader);
 					}
 				}
 			}
 		}
+		return theHandler;
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -31,7 +31,7 @@
 import org.springframework.integration.transaction.TransactionInterceptorBuilder;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.scheduling.TaskScheduler;
-import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionManager;
 import org.springframework.transaction.interceptor.TransactionInterceptor;
 import org.springframework.util.Assert;
 
@@ -112,10 +112,10 @@ public S advice(Advice... advice) {
 	 * {@code PlatformTransactionManager} and default
 	 * {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute}
 	 * for the {@link MessageHandler}.
-	 * @param transactionManager the {@link PlatformTransactionManager} to use.
+	 * @param transactionManager the {@link TransactionManager} to use.
 	 * @return the spec.
 	 */
-	public S transactional(PlatformTransactionManager transactionManager) {
+	public S transactional(TransactionManager transactionManager) {
 		return transactional(transactionManager, false);
 	}
 
@@ -124,14 +124,14 @@ public S transactional(PlatformTransactionManager transactionManager) {
 	 * {@code PlatformTransactionManager} and default
 	 * {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute}
 	 * for the {@link MessageHandler}.
-	 * @param transactionManager the {@link PlatformTransactionManager} to use.
+	 * @param transactionManager the {@link TransactionManager} to use.
 	 * @param handleMessageAdvice the flag to indicate the target {@link Advice} type:
 	 * {@code false} - regular {@link TransactionInterceptor}; {@code true} -
 	 * {@link org.springframework.integration.transaction.TransactionHandleMessageAdvice}
 	 * extension.
 	 * @return the spec.
 	 */
-	public S transactional(PlatformTransactionManager transactionManager, boolean handleMessageAdvice) {
+	public S transactional(TransactionManager transactionManager, boolean handleMessageAdvice) {
 		return transactional(new TransactionInterceptorBuilder(handleMessageAdvice)
 				.transactionManager(transactionManager)
 				.build());

spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -31,7 +31,7 @@
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.scheduling.Trigger;
-import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionManager;
 import org.springframework.transaction.interceptor.TransactionInterceptor;
 import org.springframework.util.ErrorHandler;
 
@@ -147,10 +147,10 @@ public PollerSpec advice(Advice... advice) {
 	 * provided {@code PlatformTransactionManager} and default
 	 * {@link org.springframework.transaction.interceptor.DefaultTransactionAttribute}
 	 * for the {@code pollingTask}.
-	 * @param transactionManager the {@link PlatformTransactionManager} to use.
+	 * @param transactionManager the {@link TransactionManager} to use.
 	 * @return the spec.
 	 */
-	public PollerSpec transactional(PlatformTransactionManager transactionManager) {
+	public PollerSpec transactional(TransactionManager transactionManager) {
 		return transactional(new TransactionInterceptorBuilder()
 				.transactionManager(transactionManager)
 				.build());

spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionHandleMessageAdvice.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -19,7 +19,7 @@
 import java.util.Properties;
 
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
-import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionManager;
 import org.springframework.transaction.interceptor.TransactionAttributeSource;
 import org.springframework.transaction.interceptor.TransactionInterceptor;
 
@@ -46,12 +46,16 @@ public class TransactionHandleMessageAdvice extends TransactionInterceptor imple
 	public TransactionHandleMessageAdvice() {
 	}
 
-	public TransactionHandleMessageAdvice(PlatformTransactionManager ptm, Properties attributes) {
-		super(ptm, attributes);
+	public TransactionHandleMessageAdvice(TransactionManager transactionManager, Properties transactionAttributes) {
+		setTransactionManager(transactionManager);
+		setTransactionAttributes(transactionAttributes);
 	}
 
-	public TransactionHandleMessageAdvice(PlatformTransactionManager ptm, TransactionAttributeSource tas) {
-		super(ptm, tas);
+	public TransactionHandleMessageAdvice(TransactionManager transactionManager,
+			TransactionAttributeSource transactionAttributeSource) {
+
+		setTransactionManager(transactionManager);
+		setTransactionAttributeSource(transactionAttributeSource);
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/transaction/TransactionInterceptorBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -16,7 +16,7 @@
 
 package org.springframework.integration.transaction;
 
-import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionManager;
 import org.springframework.transaction.annotation.Isolation;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
@@ -29,8 +29,8 @@
  * Provides a fluent API to build a transaction interceptor. See
  * {@link TransactionAttribute} for property meanings; if a {@link TransactionAttribute}
  * is provided, the individual properties are ignored. If a
- * {@link PlatformTransactionManager} is not provided, a single instance of
- * {@link PlatformTransactionManager} will be discovered at runtime; if you have more
+ * {@link TransactionManager} is not provided, a single instance of
+ * {@link TransactionManager} will be discovered at runtime; if you have more
  * than one transaction manager, you must inject the one you want to use here.
  * <p>
  * When the {@code handleMessageAdvice} option is in use, this builder produces
@@ -91,7 +91,7 @@ public final TransactionInterceptorBuilder transactionAttribute(TransactionAttri
 		return this;
 	}
 
-	public TransactionInterceptorBuilder transactionManager(PlatformTransactionManager transactionManager) {
+	public TransactionInterceptorBuilder transactionManager(TransactionManager transactionManager) {
 		this.transactionInterceptor.setTransactionManager(transactionManager);
 		return this;
 	}

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd

@@ -4155,56 +4155,56 @@
 	<xsd:complexType name="transactionalType">
 		<xsd:attribute name="transaction-manager" type="xsd:string" default="transactionManager">
 			<xsd:annotation>
-				<xsd:documentation><![CDATA[
-	The bean name of the PlatformTransactionManager to use.
-				]]></xsd:documentation>
+				<xsd:documentation>
+					The bean name of the PlatformTransactionManager to use.
+				</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
-						<tool:expected-type type="org.springframework.transaction.PlatformTransactionManager"/>
+						<tool:expected-type type="org.springframework.transaction.TransactionManager"/>
 					</tool:annotation>
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attribute name="propagation" default="REQUIRED">
 			<xsd:annotation>
-				<xsd:documentation source="java:org.springframework.transaction.annotation.Propagation"><![CDATA[
-	The transaction propagation behavior.
-				]]></xsd:documentation>
+				<xsd:documentation source="java:org.springframework.transaction.annotation.Propagation">
+					The transaction propagation behavior.
+				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:simpleType>
 				<xsd:union memberTypes="propagationEnumeration xsd:string"/>
 			</xsd:simpleType>
 		</xsd:attribute>
 		<xsd:attribute name="isolation" default="DEFAULT">
 			<xsd:annotation>
-				<xsd:documentation source="java:org.springframework.transaction.annotation.Isolation"><![CDATA[
-	The transaction isolation level.
-				]]></xsd:documentation>
+				<xsd:documentation source="java:org.springframework.transaction.annotation.Isolation">
+					The transaction isolation level.
+				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:simpleType>
 				<xsd:union memberTypes="isolationEnumeration xsd:string"/>
 			</xsd:simpleType>
 		</xsd:attribute>
 		<xsd:attribute name="timeout" type="xsd:string" default="-1">
 			<xsd:annotation>
-				<xsd:documentation><![CDATA[
-	The transaction timeout value (in seconds).
-				]]></xsd:documentation>
+				<xsd:documentation>
+					 The transaction timeout value (in seconds).
+				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attribute name="read-only" type="xsd:string" default="false">
 			<xsd:annotation>
-				<xsd:documentation><![CDATA[
-	Is this transaction read-only?
-				]]></xsd:documentation>
+				<xsd:documentation>
+					Is this transaction read-only?
+				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
 		<xsd:attribute name="synchronization-factory" type="xsd:string">
 			<xsd:annotation>
-				<xsd:documentation><![CDATA[
+				<xsd:documentation>
 	Reference to an instance of org.springframework.integration.transaction.TransactionSynchronizationFactory
 	which will return an instance of org.springframework.transaction.support.TransactionSynchronization via its create(..) method.
-				]]></xsd:documentation>
+				</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
 						<tool:expected-type

src/reference/asciidoc/transactions.adoc

@@ -255,3 +255,12 @@ The following example shows how to do so:
 <bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager" />
 ----
 ====
+
+[[reactive-transactions]]
+=== Reactive Transactions
+
+Starting with version 5.3, a `ReactiveTransactionManager` can also be used together with a `TransactionInterceptor` advice for endpoints returning a reactive type.
+This includes `MessageSource` and `ReactiveMessageHandler` implementations (e.g. `ReactiveMongoDbMessageSource`) which produce a message with a `Flux` or `Mono` payload.
+All other reply producing message handler implementations can rely on a `ReactiveTransactionManager` when their reply payload is also some reactive type.
+
+

src/reference/asciidoc/whats-new.adoc

@@ -57,6 +57,10 @@ A new `publishSubscribeChannel()` operator, based on the `BroadcastCapableChanne
 This fluent API has its advantage when we configure sub-flows as pub-sub subscribers for broker-backed channels like `SubscribableJmsChannel`, `SubscribableRedisChannel` etc.
 See <<./dsl.adoc#java-dsl-subflows,Sub-flows support>> for more information.
 
+Transactional support in Spring Integration now also includes options to configure a `ReactiveTransactionManager` if a `MessageSource` or `MessageHandler` implementation produces a reactive type for payload to send.
+See `TransactionInterceptorBuilder` for more information.
+See also <<./transactions.adoc#reactive-transactions,Reactive Transactions>>.
+
 [[x5.3-amqp]]
 === AMQP Changes