INT-3770: Add TX Support from Mid-flow

JIRA: https://jira.spring.io/browse/INT-3770,
https://jira.spring.io/browse/INT-4107

Having `TransactionHandleMessageAdvice` we can start TX from any `MessageHandler.handleMessage()`

* Add `<transactional>` alongside with the `<request-handler-advice-chain>` for those components which produce reply
* Merge `<transactional>` and `<request-handler-advice-chain>` configuration to a single `ManagedList`
* Rework JPA `<transactional>` in favor of common solution
* Some polishing and refactoring

AbstractPollingEndpoint: avoid `new ArrayList` if we don't  have `receiveOnlyAdvice`s

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

@@ -29,6 +29,7 @@ import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.ManagedSet;
 import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
@@ -86,11 +87,18 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, "output-channel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(handlerBuilder, element, "order");
 
+		Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
 		Element adviceChainElement = DomUtils.getChildElementByTagName(element,
 				IntegrationNamespaceUtils.REQUEST_HANDLER_ADVICE_CHAIN);
-		IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, null,
+
+		@SuppressWarnings("rawtypes")
+		ManagedList adviceChain = IntegrationNamespaceUtils.configureAdviceChain(adviceChainElement, txElement, true,
 				handlerBuilder.getRawBeanDefinition(), parserContext);
 
+		if (!CollectionUtils.isEmpty(adviceChain)) {
+			handlerBuilder.addPropertyValue("adviceChain", adviceChain);
+		}
+
 		AbstractBeanDefinition handlerBeanDefinition = handlerBuilder.getBeanDefinition();
 		String inputChannelAttributeName = this.getInputChannelAttributeName();
 		boolean hasInputChannelAttribute = element.hasAttribute(inputChannelAttributeName);
@@ -121,6 +129,10 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
 
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ConsumerEndpointFactoryBean.class);
 
+		if (!CollectionUtils.isEmpty(adviceChain)) {
+			builder.addPropertyValue("adviceChain", adviceChain);
+		}
+
 		String handlerBeanName = BeanDefinitionReaderUtils.generateBeanName(handlerBeanDefinition, parserContext.getRegistry());
 		String[] handlerAlias = IntegrationNamespaceUtils.generateAlias(element);
 		parserContext.registerBeanComponent(new BeanComponentDefinition(handlerBeanDefinition, handlerBeanName, handlerAlias));

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

@@ -27,6 +27,7 @@ import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 
@@ -81,12 +82,14 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 
 	private void configureRequestHandlerAdviceChain(Element element, ParserContext parserContext,
 							BeanDefinition handlerBeanDefinition, BeanDefinitionBuilder consumerBuilder) {
+		Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
 		Element adviceChainElement = DomUtils.getChildElementByTagName(element,
 				IntegrationNamespaceUtils.REQUEST_HANDLER_ADVICE_CHAIN);
 		@SuppressWarnings("rawtypes")
 		ManagedList adviceChain =
-				IntegrationNamespaceUtils.configureAdviceChain(adviceChainElement, null, handlerBeanDefinition, parserContext);
-		if (adviceChain != null) {
+				IntegrationNamespaceUtils.configureAdviceChain(adviceChainElement, txElement, handlerBeanDefinition,
+						parserContext);
+		if (!CollectionUtils.isEmpty(adviceChain)) {
 			/*
 			 * For ARPMH, the advice chain is injected so just the handleRequestMessage method is advised.
 			 * Sometime ARPMHs do double duty as a gateway and a channel adapter. The parser subclass

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

@@ -101,6 +101,10 @@ public class DelayerParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, txElement,
 				builder.getRawBeanDefinition(), parserContext, "delayedAdviceChain");
 
+		if (txElement != null) {
+			element.removeChild(txElement);
+		}
+
 		return builder;
 	}
 

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

@@ -48,10 +48,12 @@ import org.springframework.integration.config.FixedSubscriberChannelBeanFactoryP
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.endpoint.AbstractPollingEndpoint;
+import org.springframework.integration.transaction.TransactionHandleMessageAdvice;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
 import org.springframework.transaction.interceptor.TransactionInterceptor;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 
@@ -381,19 +383,34 @@ public abstract class IntegrationNamespaceUtils {
 	 * Parse a "transactional" element and configure a {@link TransactionInterceptor}
 	 * with "transactionManager" and other "transactionDefinition" properties.
 	 * For example, this advisor will be applied on the Polling Task proxy.
-	 *
 	 * @param txElement The transactional element.
 	 * @return The bean definition.
-	 *
 	 * @see AbstractPollingEndpoint
 	 */
 	public static BeanDefinition configureTransactionAttributes(Element txElement) {
+		return configureTransactionAttributes(txElement, false);
+	}
+
+	/**
+	 * Parse a "transactional" element and configure a {@link TransactionInterceptor}
+	 * or {@link TransactionHandleMessageAdvice}
+	 * with "transactionManager" and other "transactionDefinition" properties.
+	 * For example, this advisor will be applied on the Polling Task proxy.
+	 * @param txElement The transactional element.
+	 * @param handleMessageAdvice flag if to use {@link TransactionHandleMessageAdvice}
+	 * or regular {@link TransactionInterceptor}
+	 * @return The bean definition.
+	 * @see AbstractPollingEndpoint
+	 */
+	public static BeanDefinition configureTransactionAttributes(Element txElement, boolean handleMessageAdvice) {
 		BeanDefinition txDefinition = configureTransactionDefinition(txElement);
 		BeanDefinitionBuilder attributeSourceBuilder =
 				BeanDefinitionBuilder.genericBeanDefinition(MatchAlwaysTransactionAttributeSource.class);
 		attributeSourceBuilder.addPropertyValue("transactionAttribute", txDefinition);
 		BeanDefinitionBuilder txInterceptorBuilder =
-				BeanDefinitionBuilder.genericBeanDefinition(TransactionInterceptor.class);
+				BeanDefinitionBuilder.genericBeanDefinition(handleMessageAdvice
+						? TransactionHandleMessageAdvice.class
+						: TransactionInterceptor.class);
 		txInterceptorBuilder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
 		txInterceptorBuilder.addPropertyValue("transactionAttributeSource", attributeSourceBuilder.getBeanDefinition());
 		return txInterceptorBuilder.getBeanDefinition();
@@ -426,31 +443,47 @@ public abstract class IntegrationNamespaceUtils {
 
 	public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
 			BeanDefinition parentBeanDefinition, ParserContext parserContext) {
-		configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, parentBeanDefinition, parserContext,
-				"adviceChain");
+		configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, false, parentBeanDefinition, parserContext);
+	}
+
+	public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement,
+			Element txElement, boolean handleMessageAdvice, BeanDefinition parentBeanDefinition,
+			ParserContext parserContext) {
+		configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, handleMessageAdvice,
+				parentBeanDefinition, parserContext, "adviceChain");
+	}
+
+	public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
+			BeanDefinition parentBeanDefinition, ParserContext parserContext, String propertyName) {
+		configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, false, parentBeanDefinition,
+				parserContext, propertyName);
 	}
 
 	@SuppressWarnings({ "rawtypes" })
 	public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
-			BeanDefinition parentBeanDefinition, ParserContext parserContext, String propertyName) {
-		ManagedList adviceChain = configureAdviceChain(adviceChainElement, txElement, parentBeanDefinition,
-				parserContext);
-		if (adviceChain != null) {
+			boolean handleMessageAdvice, BeanDefinition parentBeanDefinition, ParserContext parserContext,
+			String propertyName) {
+		ManagedList adviceChain = configureAdviceChain(adviceChainElement, txElement, handleMessageAdvice,
+				parentBeanDefinition, parserContext);
+		if (!CollectionUtils.isEmpty(adviceChain)) {
 			parentBeanDefinition.getPropertyValues().add(propertyName, adviceChain);
 		}
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@SuppressWarnings("rawtypes")
 	public static ManagedList configureAdviceChain(Element adviceChainElement, Element txElement,
 			BeanDefinition parentBeanDefinition, ParserContext parserContext) {
-		ManagedList adviceChain = null;
-		// Schema validation ensures txElement and adviceChainElement are mutually exclusive
+		return configureAdviceChain(adviceChainElement, txElement, false, parentBeanDefinition, parserContext);
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	public static ManagedList configureAdviceChain(Element adviceChainElement, Element txElement,
+			boolean handleMessageAdvice, BeanDefinition parentBeanDefinition, ParserContext parserContext) {
+		ManagedList adviceChain = new ManagedList();
 		if (txElement != null) {
-			adviceChain = new ManagedList();
-			adviceChain.add(IntegrationNamespaceUtils.configureTransactionAttributes(txElement));
+			adviceChain.add(configureTransactionAttributes(txElement, handleMessageAdvice));
 		}
 		if (adviceChainElement != null) {
-			adviceChain = new ManagedList();
 			NodeList childNodes = adviceChainElement.getChildNodes();
 			for (int i = 0; i < childNodes.getLength(); i++) {
 				Node child = childNodes.item(i);

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java

@@ -16,12 +16,12 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledFuture;
+import java.util.stream.Collectors;
 
 import org.aopalliance.aop.Advice;
 
@@ -174,30 +174,26 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 
 	@SuppressWarnings("unchecked")
 	private Runnable createPoller() throws Exception {
-		List<Advice> receiveOnlyAdviceChain = new ArrayList<Advice>();
+		List<Advice> receiveOnlyAdviceChain = null;
 		if (!CollectionUtils.isEmpty(this.adviceChain)) {
-			for (Advice advice : this.adviceChain) {
-				if (isReceiveOnlyAdvice(advice)) {
-					receiveOnlyAdviceChain.add(advice);
-				}
-			}
+			receiveOnlyAdviceChain = this.adviceChain.stream()
+					.filter(this::isReceiveOnlyAdvice)
+					.collect(Collectors.toList());
 		}
 
-		Callable<Boolean> pollingTask = () -> doPoll();
+		Callable<Boolean> pollingTask = this::doPoll;
 
 		List<Advice> adviceChain = this.adviceChain;
 		if (!CollectionUtils.isEmpty(adviceChain)) {
 			ProxyFactory proxyFactory = new ProxyFactory(pollingTask);
 			if (!CollectionUtils.isEmpty(adviceChain)) {
-				for (Advice advice : adviceChain) {
-					if (!isReceiveOnlyAdvice(advice)) {
-						proxyFactory.addAdvice(advice);
-					}
-				}
+				adviceChain.stream()
+						.filter(advice -> !isReceiveOnlyAdvice(advice))
+						.forEach(proxyFactory::addAdvice);
 			}
 			pollingTask = (Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader);
 		}
-		if (receiveOnlyAdviceChain.size() > 0) {
+		if (receiveOnlyAdviceChain != null) {
 			applyReceiveOnlyAdviceChain(receiveOnlyAdviceChain);
 		}
 		return new Poller(pollingTask);

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -132,9 +132,10 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 	@Override
 	protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
 		if (AopUtils.isAopProxy(this.source)) {
-			this.appliedAdvices.forEach(((Advised) this.source)::removeAdvice);
+			Advised source = (Advised) this.source;
+			this.appliedAdvices.forEach(source::removeAdvice);
 			for (Advice advice : chain) {
-				((Advised) this.source).addAdvisor(adviceToReceiveAdvisor(advice));
+				source.addAdvisor(adviceToReceiveAdvisor(advice));
 			}
 		}
 		else {

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

@@ -1367,6 +1367,7 @@
 					</xsd:complexContent>
 				</xsd:complexType>
 			</xsd:element>
+			<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
 			<xsd:element name="request-handler-advice-chain" type="handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
 		</xsd:sequence>
 		<xsd:attribute name="request-channel" type="xsd:string" use="optional">
@@ -1665,6 +1666,7 @@
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:choice minOccurs="0" maxOccurs="2">
+				<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
 				<xsd:element name="request-handler-advice-chain" type="handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
 				<xsd:element ref="poller" />
 			</xsd:choice>
@@ -2871,6 +2873,7 @@
 		<xsd:complexContent>
 			<xsd:extension base="expressionOrInnerEndpointDefinitionAwareNoAdviceChain">
 			    <xsd:sequence>
+					<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
 					<xsd:element name="request-handler-advice-chain" minOccurs="0" maxOccurs="1">
 						<xsd:complexType>
 							<xsd:complexContent>
@@ -4125,6 +4128,7 @@
 				<xsd:choice minOccurs="0" maxOccurs="3">
 					<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
 					<xsd:element name="expression" type="innerExpressionType" minOccurs="0" maxOccurs="1" />
+					<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
 					<xsd:element name="request-handler-advice-chain" type="handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
 					<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="1" />
 				</xsd:choice>

spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml

@@ -27,6 +27,9 @@
 				input-channel="aggregatorWithCustomMGPReferenceInput" output-channel="outputChannel"/>
 
 	<channel id="completelyDefinedAggregatorInput"/>
+
+	<beans:bean id="transactionManager" class="org.springframework.integration.transaction.PseudoTransactionManager"/>
+
 	<aggregator id="completelyDefinedAggregator"
 		input-channel="completelyDefinedAggregatorInput"
 		output-channel="outputChannel"
@@ -44,7 +47,7 @@
 		scheduler="scheduler"
 		message-store="store"
 		order="5">
-			<expire-advice-chain/>
+			<expire-transactional/>
 	</aggregator>
 
 	<beans:bean id="lockRegistry" class="org.springframework.integration.support.locks.DefaultLockRegistry"/>
@@ -116,4 +119,5 @@
 		class="org.springframework.integration.config.MaxValueReleaseStrategy">
 		<beans:constructor-arg value="10" />
 	</beans:bean>
+
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java

@@ -34,11 +34,12 @@ import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Assert;
-import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -63,6 +64,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Marius Bogoevici
@@ -73,15 +75,12 @@ import org.springframework.messaging.SubscribableChannel;
  * @author Gunnar Hillert
  * @author Gary Russell
  */
+@RunWith(SpringRunner.class)
 public class AggregatorParserTests {
 
+	@Autowired
 	private ApplicationContext context;
 
-	@Before
-	public void setUp() {
-		this.context = new ClassPathXmlApplicationContext("aggregatorParserTests.xml", this.getClass());
-	}
-
 	@Test
 	public void testAggregation() {
 		MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceInput");
@@ -90,11 +89,11 @@ public class AggregatorParserTests {
 		outboundMessages.add(createMessage("123", "id1", 3, 1, null));
 		outboundMessages.add(createMessage("789", "id1", 3, 3, null));
 		outboundMessages.add(createMessage("456", "id1", 3, 2, null));
-		for (Message<?> message : outboundMessages) {
-			input.send(message);
-		}
-		assertEquals("One and only one message must have been aggregated", 1, aggregatorBean.getAggregatedMessages()
-				.size());
+
+		outboundMessages.forEach(input::send);
+
+		assertEquals("One and only one message must have been aggregated", 1,
+				aggregatorBean.getAggregatedMessages().size());
 		Message<?> aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1");
 		assertEquals("The aggregated message payload is not correct", "123456789", aggregatedMessage.getPayload());
 		Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
@@ -111,9 +110,9 @@ public class AggregatorParserTests {
 		outboundMessages.add(createMessage("123", "id1", 3, 1, null));
 		outboundMessages.add(createMessage("789", "id1", 3, 3, null));
 		outboundMessages.add(createMessage("456", "id1", 3, 2, null));
-		for (Message<?> message : outboundMessages) {
-			input.send(message);
-		}
+
+		outboundMessages.forEach(input::send);
+
 		assertEquals(3, output.getQueueSize());
 		output.purge(null);
 	}
@@ -127,7 +126,9 @@ public class AggregatorParserTests {
 		outboundMessages.add(createMessage("123", "id1", 3, 1, null));
 		outboundMessages.add(createMessage("789", "id1", 3, 3, null));
 		outboundMessages.add(createMessage("456", "id1", 3, 2, null));
+
 		outboundMessages.forEach(input::send);
+
 		assertEquals(3, output.getQueueSize());
 		output.purge(null);
 	}
@@ -142,7 +143,9 @@ public class AggregatorParserTests {
 		outboundMessages.add(MessageBuilder.withPayload("123").setHeader("foo", "1").build());
 		outboundMessages.add(MessageBuilder.withPayload("456").setHeader("foo", "1").build());
 		outboundMessages.add(MessageBuilder.withPayload("789").setHeader("foo", "1").build());
+
 		outboundMessages.forEach(input::send);
+
 		assertEquals("The aggregated message payload is not correct", "[123]", aggregatedMessage.get().getPayload()
 				.toString());
 		Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);