diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index b1ed21ff00..84cb5588ac 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/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 channelResolver; + private DestinationResolver 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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java index 8c48a04ec0..d16066ea89 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BarrierSpec.java +++ b/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 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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java index 9d2cbf00c2..ed41e11c2f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java @@ -267,7 +267,7 @@ public abstract class ConsumerEndpointSpec, @Override protected Tuple2 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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 758e391ee5..431421448b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/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 adviceChain = new LinkedList<>(); + + private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); + + private boolean requiresReply = false; + private volatile RequestHandler advisedRequestHandler; - private volatile List 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 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: diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java index dafbb2ac06..53a0903286 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java +++ b/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()); + } + } } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TwitterSearchOutboundGatewayParserTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TwitterSearchOutboundGatewayParserTests.java index 74ac5c01e0..99aa3272c5 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TwitterSearchOutboundGatewayParserTests.java +++ b/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)); } diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 58411e5e40..57dc4e6ff4 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/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