INT-4487: Allow reconfiguring ARPMH.adviceChain

JIRA: https://jira.spring.io/browse/INT-4487

Currently the `AbstractReplyProducingMessageHandler` applies advices
only in the `afterPropertiesSet()` phase.
When we use bean references for `MessageHandler`s in Java DSL, the
`.advice()` provided in the flow definition is not applied to the
`AbstractReplyProducingMessageHandler`.

* For consistency with the endpoint configured by the DSL definition,
override and reconfigure advices in the `AbstractReplyProducingMessageHandler`
bean reference during DSL parsing and processing.

Fix `TwitterSearchOutboundGatewayParserTests` for the proper assert

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -75,26 +75,28 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 
 	private final ConversionService defaultConversionService = DefaultConversionService.getSharedInstance();
 
-	private volatile DestinationResolver<MessageChannel> channelResolver;
+	private DestinationResolver<MessageChannel> channelResolver;
 
-	private volatile String beanName;
+	private String beanName;
 
-	private volatile String componentName;
+	private String componentName;
 
-	private volatile BeanFactory beanFactory;
+	private BeanFactory beanFactory;
 
-	private volatile TaskScheduler taskScheduler;
+	private TaskScheduler taskScheduler;
 
-	private volatile Properties integrationProperties = IntegrationProperties.defaults();
+	private Properties integrationProperties = IntegrationProperties.defaults();
 
-	private volatile ConversionService conversionService;
+	private ConversionService conversionService;
 
-	private volatile ApplicationContext applicationContext;
+	private ApplicationContext applicationContext;
 
-	private volatile MessageBuilderFactory messageBuilderFactory;
+	private MessageBuilderFactory messageBuilderFactory;
 
 	private Expression expression;
 
+	private boolean initialized;
+
 	@Override
 	public final void setBeanName(String beanName) {
 		this.beanName = beanName;
@@ -181,6 +183,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 			}
 			throw new BeanInitializationException("failed to initialize", e);
 		}
+
+		this.initialized = true;
 	}
 
 	/**
@@ -190,6 +194,14 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
 	protected void onInit() throws Exception {
 	}
 
+	/**
+	 * Return the status of this component if it has been initialized already.
+	 * @return the flag if this component has been initialized already.
+	 */
+	protected boolean isInitialized() {
+		return this.initialized;
+	}
+
 	protected BeanFactory getBeanFactory() {
 		return this.beanFactory;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -96,7 +96,9 @@ public class BarrierSpec extends ConsumerEndpointSpec<BarrierSpec, BarrierMessag
 	@Override
 	public Tuple2<ConsumerEndpointFactoryBean, BarrierMessageHandler> doGet() {
 		this.handler = new BarrierMessageHandler(this.timeout, this.outputProcessor, this.correlationStrategy);
-		this.handler.setAdviceChain(this.adviceChain);
+		if (!this.adviceChain.isEmpty()) {
+			this.handler.setAdviceChain(this.adviceChain);
+		}
 		this.handler.setRequiresReply(this.requiresReply);
 		this.handler.setSendTimeout(this.sendTimeout);
 		this.handler.setAsync(this.async);

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

@@ -267,7 +267,7 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
 	@Override
 	protected Tuple2<ConsumerEndpointFactoryBean, H> doGet() {
 		this.endpointFactoryBean.setAdviceChain(this.adviceChain);
-		if (this.handler instanceof AbstractReplyProducingMessageHandler) {
+		if (this.handler instanceof AbstractReplyProducingMessageHandler && !this.adviceChain.isEmpty()) {
 			((AbstractReplyProducingMessageHandler) this.handler).setAdviceChain(this.adviceChain);
 		}
 		this.endpointFactoryBean.setHandler(this.handler);

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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,6 +16,7 @@
 
 package org.springframework.integration.handler;
 
+import java.util.LinkedList;
 import java.util.List;
 
 import org.aopalliance.aop.Advice;
@@ -26,7 +27,6 @@ import org.springframework.integration.handler.advice.HandleMessageAdvice;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
-import org.springframework.util.CollectionUtils;
 
 /**
  * Base class for MessageHandlers that are capable of producing replies.
@@ -41,14 +41,14 @@ import org.springframework.util.CollectionUtils;
 public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageProducingHandler
 		implements BeanClassLoaderAware {
 
+	private final List<Advice> adviceChain = new LinkedList<>();
+
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+
+	private boolean requiresReply = false;
+
 	private volatile RequestHandler advisedRequestHandler;
 
-	private volatile List<Advice> adviceChain;
-
-	private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
-
-	private volatile boolean requiresReply = false;
-
 	/**
 	 * Flag whether a reply is required. If true an incoming message MUST result in a reply message being sent.
 	 * If false an incoming message MAY result in a reply message being sent. Default is false.
@@ -62,13 +62,23 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 		return this.requiresReply;
 	}
 
+	/**
+	 * Configure a list of {@link Advice}s to proxy a {@link #handleRequestMessage(Message)} method.
+	 * @param adviceChain the list of {@link Advice}s to use.
+	 */
 	public void setAdviceChain(List<Advice> adviceChain) {
-		Assert.notNull(adviceChain, "adviceChain cannot be null");
-		this.adviceChain = adviceChain;
+		Assert.notEmpty(adviceChain, "adviceChain cannot be empty");
+		synchronized (this.adviceChain) {
+			this.adviceChain.clear();
+			this.adviceChain.addAll(adviceChain);
+			if (isInitialized()) {
+				initAdvisedRequestHandlerIfAny();
+			}
+		}
 	}
 
 	protected boolean hasAdviceChain() {
-		return this.adviceChain != null && this.adviceChain.size() > 0;
+		return this.adviceChain.size() > 0;
 	}
 
 	@Override
@@ -80,7 +90,12 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 	@Override
 	protected final void onInit() throws Exception {
 		super.onInit();
-		if (!CollectionUtils.isEmpty(this.adviceChain)) {
+		initAdvisedRequestHandlerIfAny();
+		doInit();
+	}
+
+	private void initAdvisedRequestHandlerIfAny() {
+		if (!this.adviceChain.isEmpty()) {
 			ProxyFactory proxyFactory = new ProxyFactory(new AdvisedRequestHandler());
 			boolean advised = false;
 			for (Advice advice : this.adviceChain) {
@@ -93,7 +108,6 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 				this.advisedRequestHandler = (RequestHandler) proxyFactory.getProxy(this.beanClassLoader);
 			}
 		}
-		doInit();
 	}
 
 	protected void doInit() {
@@ -154,9 +168,6 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 
 		Object handleRequestMessage(Message<?> requestMessage);
 
-		@Override
-		String toString();
-
 		/**
 		 * Utility method, intended for use in message handler advice classes to get
 		 * information about the advised object. For example:

spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java

@@ -25,7 +25,9 @@ import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.aopalliance.intercept.MethodInterceptor;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -37,11 +39,12 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Transformers;
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
-import org.springframework.integration.dsl.context.IntegrationFlowContext.IntegrationFlowRegistration;
+import org.springframework.integration.ip.tcp.TcpOutboundGateway;
 import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
 import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -60,6 +63,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.0
  *
@@ -71,6 +75,9 @@ public class IpIntegrationTests {
 	@Autowired
 	private AbstractServerConnectionFactory server1;
 
+	@Autowired
+	private AbstractClientConnectionFactory client1;
+
 	@Autowired
 	private IntegrationFlowContext flowContext;
 
@@ -78,6 +85,10 @@ public class IpIntegrationTests {
 	@Qualifier("outUdpAdapter.input")
 	private MessageChannel udpOut;
 
+	@Autowired
+	@Qualifier("clientTcpFlow.input")
+	private MessageChannel clientTcpFlowInput;
+
 	@Autowired
 	private UnicastReceivingChannelAdapter udpInbound;
 
@@ -87,8 +98,11 @@ public class IpIntegrationTests {
 	@Autowired
 	private Config config;
 
+	@Autowired
+	private AtomicBoolean adviceCalled;
+
 	@Test
-	public void testTcpAdapters() throws Exception {
+	public void testTcpAdapters() {
 		ApplicationEventPublisher publisher = e -> { };
 		AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).id("server").get();
 		assertEquals("server", server.getComponentName());
@@ -117,15 +131,15 @@ public class IpIntegrationTests {
 	@Test
 	public void testTcpGateways() {
 		TestingUtilities.waitListening(this.server1, null);
-		IntegrationFlow flow = f -> f
-				.handle(Tcp.outboundGateway(Tcp.netClient("localhost", this.server1.getPort())
-						.serializer(TcpCodecs.crlf())
-						.deserializer(TcpCodecs.lengthHeader1())
-						.id("client1"))
-					.remoteTimeout(m -> 5000))
-				.transform(Transformers.objectToString());
-		IntegrationFlowRegistration theFlow = this.flowContext.registration(flow).register();
-		assertThat(theFlow.getMessagingTemplate().convertSendAndReceive("foo", String.class), equalTo("FOO"));
+		this.client1.stop();
+		this.client1.setPort(this.server1.getPort());
+		this.client1.start();
+
+		MessagingTemplate messagingTemplate = new MessagingTemplate(this.clientTcpFlowInput);
+
+		assertThat(messagingTemplate.convertSendAndReceive("foo", String.class), equalTo("FOO"));
+
+		assertTrue(this.adviceCalled.get());
 	}
 
 	@Test
@@ -204,6 +218,41 @@ public class IpIntegrationTests {
 			};
 		}
 
+		@Bean
+		public AbstractClientConnectionFactory client1() {
+			return Tcp.netClient("localhost", server1().getPort())
+					.serializer(TcpCodecs.crlf())
+					.deserializer(TcpCodecs.lengthHeader1())
+					.get();
+		}
+
+		@Bean
+		public TcpOutboundGateway tcpOut() {
+			return Tcp.outboundGateway(client1())
+					.remoteTimeout(m -> 5000)
+					.get();
+		}
+
+		@Bean
+		public AtomicBoolean adviceCalled() {
+			return new AtomicBoolean();
+		}
+
+		@Bean
+		public MethodInterceptor testAdvice() {
+			return invocation -> {
+				adviceCalled().set(true);
+				return invocation.proceed();
+			};
+		}
+
+		@Bean
+		public IntegrationFlow clientTcpFlow() {
+			return f -> f
+					.handle(tcpOut(), e -> e.advice(testAdvice()))
+					.transform(Transformers.objectToString());
+		}
+
 	}
 
 }

spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TwitterSearchOutboundGatewayParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -20,7 +20,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 
-import java.util.ArrayList;
+import java.util.List;
 
 import org.hamcrest.Matchers;
 import org.junit.Test;
@@ -39,6 +39,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
@@ -75,7 +77,7 @@ public class TwitterSearchOutboundGatewayParserTests {
 	@Test
 	public void testAdvised() {
 		assertSame(twitter, TestUtils.getPropertyValue(polledAndAdvisedTSOG, "handler.twitter"));
-		assertThat(TestUtils.getPropertyValue(polledAndAdvisedTSOG, "handler.adviceChain", ArrayList.class).get(0),
+		assertThat(TestUtils.getPropertyValue(polledAndAdvisedTSOG, "handler.adviceChain", List.class).get(0),
 				Matchers.instanceOf(RequestHandlerRetryAdvice.class));
 	}
 

src/reference/asciidoc/dsl.adoc

@@ -236,6 +236,28 @@ public IntegrationFlow flow2() {
 
 In addition the `EndpointSpec` provides an `id()` method to allow you to register an endpoint bean with a  given bean name, rather than a generated one.
 
+If the `MessageHandler` is referenced as a bean, then any existing `adviceChain` configuration will be overridden if the `.advice()` method is present in the DSL definition:
+
+[source,java]
+----
+@Bean
+public TcpOutboundGateway tcpOut() {
+    TcpOutboundGateway gateway = new TcpOutboundGateway();
+    gateway.setConnectionFactory(cf());
+    gateway.setAdviceChain(Collections.singletonList(fooAdvice()));
+    return gateway;
+}
+
+@Bean
+public IntegrationFlow clientTcpFlow() {
+    return f -> f
+        .handle(tcpOut(), e -> e.advice(testAdvice()))
+        .transform(Transformers.objectToString());
+}
+----
+
+i.e. they are not merged, only the `testAdvice()` bean is used in this case.
+
 [[java-dsl-transformers]]
 === Transformers