diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 28e806a47a..f19269806c 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 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. @@ -52,7 +52,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel private final String channelName; - private volatile Queue queue; + private Queue queue; private volatile int executorInterceptorsSize; @@ -188,30 +188,33 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } } Object object = performReceive(timeout); + Message message = null; if (object == null) { if (isLoggingEnabled() && logger.isTraceEnabled()) { logger.trace("postReceive on channel '" + this + "', message is null"); } - return null; - } - if (countsEnabled) { - getMetrics().afterReceive(); - counted = true; - } - Message message; - if (object instanceof Message) { - message = (Message) object; } else { - message = getMessageBuilderFactory() - .withPayload(object) - .build(); - } - if (isLoggingEnabled() && logger.isDebugEnabled()) { - logger.debug("postReceive on channel '" + this + "', message: " + message); + if (countsEnabled) { + getMetrics().afterReceive(); + counted = true; + } + if (object instanceof Message) { + message = (Message) object; + } + else { + message = getMessageBuilderFactory() + .withPayload(object) + .build(); + } + if (isLoggingEnabled() && logger.isDebugEnabled()) { + logger.debug("postReceive on channel '" + this + "', message: " + message); + } } if (interceptorStack != null) { - message = interceptorList.postReceive(message, this); + if (message != null) { + message = interceptorList.postReceive(message, this); + } interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); } return message; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index a58b6ce77f..51edfe3663 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -108,7 +108,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel return null; } } - Message message = this.doReceive(timeout); + Message message = doReceive(timeout); if (countsEnabled && message != null) { if (getMetricsCaptor() != null) { incrementReceiveCounter(); @@ -123,7 +123,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel logger.trace("postReceive on channel '" + this + "', message is null"); } if (!CollectionUtils.isEmpty(interceptorStack)) { - message = interceptorList.postReceive(message, this); + if (message != null) { + message = interceptorList.postReceive(message, this); + } interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); } return message; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java index 9906d9b8bb..90210f6cc2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java @@ -208,7 +208,6 @@ public class ChannelInterceptorTests { @Test public void testPostReceiveInterceptor() { - final AtomicInteger invokedCount = new AtomicInteger(); final AtomicInteger messageCount = new AtomicInteger(); channel.addInterceptor(new ChannelInterceptor() { @@ -216,21 +215,16 @@ public class ChannelInterceptorTests { public Message postReceive(Message message, MessageChannel channel) { assertNotNull(channel); assertSame(ChannelInterceptorTests.this.channel, channel); - if (message != null) { - messageCount.incrementAndGet(); - } - invokedCount.incrementAndGet(); + messageCount.incrementAndGet(); return message; } }); channel.receive(0); - assertEquals(1, invokedCount.get()); assertEquals(0, messageCount.get()); channel.send(new GenericMessage("test")); Message result = channel.receive(0); assertNotNull(result); - assertEquals(2, invokedCount.get()); assertEquals(1, messageCount.get()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java index 3a64b34b11..34bd2fdf59 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java @@ -155,7 +155,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceSuperclassAnnotatedMethod() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelFoo", DirectChannel.class); final Method fooMethod = Foo.class.getMethod("foo", String.class); final AtomicBoolean called = new AtomicBoolean(); @@ -180,7 +181,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceSuperclassAnnotatedMethodOverridePE() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests2-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests2-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelFoo", DirectChannel.class); final Method fooMethod = Foo.class.getMethod("foo", String.class); final AtomicBoolean called = new AtomicBoolean(); @@ -202,7 +204,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceAnnotatedMethod() { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBar", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -214,17 +217,17 @@ public class GatewayInterfaceTests { @Test public void testWithServiceSuperclassUnAnnotatedMethod() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); final Method bazMethod = Foo.class.getMethod("baz", String.class); final AtomicBoolean called = new AtomicBoolean(); MessageHandler handler = message -> { - assertThat((String) message.getHeaders().get("name"), equalTo("overrideGlobal")); - assertThat( - (String) message.getHeaders().get("string"), + assertThat(message.getHeaders().get("name"), equalTo("overrideGlobal")); + assertThat(message.getHeaders().get("string"), equalTo("public abstract void org.springframework.integration.gateway.GatewayInterfaceTests$Foo.baz(java.lang.String)")); - assertThat((Method) message.getHeaders().get("object"), equalTo(bazMethod)); - assertThat((String) message.getPayload(), equalTo("hello")); + assertThat(message.getHeaders().get("object"), equalTo(bazMethod)); + assertThat(message.getPayload(), equalTo("hello")); called.set(true); }; channel.subscribe(handler); @@ -236,17 +239,17 @@ public class GatewayInterfaceTests { @Test public void testWithServiceUnAnnotatedMethodGlobalHeaderDoesntOverride() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); final Method quxMethod = Bar.class.getMethod("qux", String.class, String.class); final AtomicBoolean called = new AtomicBoolean(); MessageHandler handler = message -> { - assertThat((String) message.getHeaders().get("name"), equalTo("arg1")); - assertThat( - (String) message.getHeaders().get("string"), + assertThat(message.getHeaders().get("name"), equalTo("arg1")); + assertThat(message.getHeaders().get("string"), equalTo("public abstract void org.springframework.integration.gateway.GatewayInterfaceTests$Bar.qux(java.lang.String,java.lang.String)")); - assertThat((Method) message.getHeaders().get("object"), equalTo(quxMethod)); - assertThat((String) message.getPayload(), equalTo("hello")); + assertThat(message.getHeaders().get("object"), equalTo(quxMethod)); + assertThat(message.getPayload(), equalTo("hello")); called.set(true); }; channel.subscribe(handler); @@ -258,7 +261,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceCastAsSuperclassAnnotatedMethod() { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelFoo", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -270,7 +274,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceCastAsSuperclassUnAnnotatedMethod() { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -281,8 +286,9 @@ public class GatewayInterfaceTests { } @Test - public void testWithServiceHashcode() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + public void testWithServiceHashcode() { + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -294,7 +300,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceToString() { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -306,7 +313,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceEquals() throws Exception { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -326,7 +334,8 @@ public class GatewayInterfaceTests { @Test public void testWithServiceGetClass() { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); MessageHandler handler = mock(MessageHandler.class); channel.subscribe(handler); @@ -343,7 +352,8 @@ public class GatewayInterfaceTests { @Test public void testWithCustomMapper() { - ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); + ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this + .getClass()); DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class); final AtomicBoolean called = new AtomicBoolean(); MessageHandler handler = message -> { @@ -358,14 +368,14 @@ public class GatewayInterfaceTests { } @Test - public void testLateReply() throws Exception { + public void testLateReply() { ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass()); Bar baz = ac.getBean(Bar.class); String reply = baz.lateReply("hello", 1000, 0); assertNull(reply); PollableChannel errorChannel = ac.getBean("errorChannel", PollableChannel.class); - Message receive = errorChannel.receive(5000); + Message receive = errorChannel.receive(10000); assertNotNull(receive); MessagingException messagingException = (MessagingException) receive.getPayload(); assertThat(messagingException.getMessage(), @@ -375,7 +385,7 @@ public class GatewayInterfaceTests { @Test public void testInt2634() { - Map param = Collections.singletonMap(1, 1); + Map param = Collections.singletonMap(1, 1); Object result = this.int2634Gateway.test2(param); assertEquals(param, result); @@ -405,7 +415,7 @@ public class GatewayInterfaceTests { ListenableFuture result2 = this.execGateway.test2(Thread.currentThread()); final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference thread = new AtomicReference(); + final AtomicReference thread = new AtomicReference<>(); result2.addCallback(new ListenableFutureCallback() { @Override @@ -528,22 +538,26 @@ public class GatewayInterfaceTests { void bar(String payload); void qux(String payload, @Header("name") String nameHeader); + } public static class NotAnInterface { public void fail(String payload) { } + } public interface Baz { void baz(String payload); + } public interface NoArgumentsGateway { String pullData(); + } public static class BazMapper implements MethodArgsMessageMapper { @@ -601,7 +615,7 @@ public class GatewayInterfaceTests { Object payload; if (Thread.currentThread().equals(message.getPayload())) { // running on calling thread - need to return a Future. - payload = new AsyncResult(Thread.currentThread()); + payload = new AsyncResult<>(Thread.currentThread()); } else { payload = Thread.currentThread(); @@ -633,6 +647,7 @@ public class GatewayInterfaceTests { public GatewayProxyFactoryBean annotationGatewayProxyFactoryBean() { return new AnnotationGatewayProxyFactoryBean(GatewayByAnnotationGPFB.class); } + } @MessagingGateway diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java index b54509fa79..3c50936c81 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 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. @@ -33,12 +33,13 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * * @since 2.0 */ public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware { - private volatile String messageSelector; + private String messageSelector; private volatile int executorInterceptorsSize; @@ -95,28 +96,33 @@ public class PollableJmsChannel extends AbstractJmsChannel object = getJmsTemplate().receiveSelectedAndConvert(this.messageSelector); } + Message message = null; if (object == null) { if (logger.isTraceEnabled()) { logger.trace("postReceive on channel '" + this + "', message is null"); } - return null; - } - if (countsEnabled) { - getMetrics().afterReceive(); - counted = true; - } - Message message = null; - if (object instanceof Message) { - message = (Message) object; } else { - message = getMessageBuilderFactory().withPayload(object).build(); - } - if (logger.isDebugEnabled()) { - logger.debug("postReceive on channel '" + this + "', message: " + message); + if (countsEnabled) { + getMetrics().afterReceive(); + counted = true; + } + if (object instanceof Message) { + message = (Message) object; + } + else { + message = getMessageBuilderFactory() + .withPayload(object) + .build(); + } + if (logger.isDebugEnabled()) { + logger.debug("postReceive on channel '" + this + "', message: " + message); + } } if (interceptorStack != null) { - message = interceptorList.postReceive(message, this); + if (message != null) { + message = interceptorList.postReceive(message, this); + } interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); } return message; @@ -136,7 +142,7 @@ public class PollableJmsChannel extends AbstractJmsChannel public Message receive(long timeout) { try { DynamicJmsTemplateProperties.setReceiveTimeout(timeout); - return this.receive(); + return receive(); } finally { DynamicJmsTemplateProperties.clearReceiveTimeout(); diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index 36fdda6d5c..9da01f65a1 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/src/reference/asciidoc/channel.adoc @@ -273,7 +273,7 @@ Also, the `preReceive` method can return `false` to prevent the receive operatio NOTE: Keep in mind that `receive()` calls are only relevant for `PollableChannels`. In fact, the `SubscribableChannel` interface does not even define a `receive()` method. -The reason for this is that when a `Message` is sent to a `SubscribableChannel`, it is sent directly to one or more subscribers, depending on the type of channel (for example, +The reason for this is that when a `Message` is sent to a `SubscribableChannel`, it is sent directly to zero or more subscribers, depending on the type of channel (for example, a `PublishSubscribeChannel` sends to all of its subscribers). Therefore, the `preReceive(...)`, `postReceive(...)`, and `afterReceiveCompletion(...)` interceptor methods are invoked only when the interceptor is applied to a `PollableChannel`. @@ -282,25 +282,7 @@ It is a simple interceptor that sends the `Message` to another channel without o It can be very useful for debugging and monitoring. An example is shown in <>. -Because it is rarely necessary to implement all of the interceptor methods, a `ChannelInterceptorAdapter` class is also available for sub-classing. -It provides no-op methods (the `void` method is empty, the `Message`-returning methods return the `Message` as-is, and the `boolean` method returns `true`). -Therefore, it is often easiest to extend that class and just implement the methods that you need, as the following example shows: - -==== -[source,java] ----- -public class CountingChannelInterceptor extends ChannelInterceptorAdapter { - - private final AtomicInteger sendCount = new AtomicInteger(); - - @Override - public Message preSend(Message message, MessageChannel channel) { - sendCount.incrementAndGet(); - return message; - } -} ----- -==== +Because it is rarely necessary to implement all of the interceptor methods, the interface provides no-op methods (methods returning `void` method have no code, the `Message`-returning methods return the `Message` as-is, and the `boolean` method returns `true`). TIP: The order of invocation for the interceptor methods depends on the type of channel. As described earlier, the queue-based channels are the only ones where the receive method is intercepted in the first place. @@ -318,6 +300,11 @@ Note that the channel invokes these methods on the `ChannelInterceptor` list in Starting with version 5.1, global channel interceptors now apply to dynamically registered channels - such as through beans that are initialized by using `beanFactory.initializeBean()` or `IntegrationFlowContext` when using the Java DSL. Previously, interceptors were not applied when beans were created after the application context was refreshed. +Also, starting with version 5.1, `ChannelInterceptor.postReceive()` is no longer called when no message is received; it is no longer necessary to check for a `null` `Message`. +Previously, the method was called. +If you have an interceptor that relies on the previous behavior, implement `afterReceiveCompleted()` instead, since that method is invoked, regardless of whether a message is received or not. + + [[channel-template]] ==== `MessagingTemplate` diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index f9d5b5c5c3..a3ee07bddd 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -62,6 +62,14 @@ Previously: Global channel interceptors now apply to dynamically registered channels, such as through the `IntegrationFlowContext` when using the Java DSL or beans that are initialized using `beanFactory.initializeBean()`. Previously, when beans were created after the application context was refreshed, interceptors were not applied. +[[x5.1-channel-interceptors]] +==== Channel Interceptors + +`ChannelInterceptor.postReceive()` is no longer called when no message is received; it is no longer necessary to check for a `null` `Message`. +Previously, the method was called. +If you have an interceptor that relies on the previous behavior, implement `afterReceiveCompleted()` instead, since that method is invoked, regardless of whether a message is received or not. +Furthermore, the `PolledAmqpChannel` and `PolledJmsChannel` previously did not invoke `afterReceiveCompleted()` with `null`; they now do. + [[x5.1-object-to-json-transformer]] ==== `ObjectToJsonTransformer`