Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4526: Fix Channel Interceptor NonNullApi Call

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4526

- `Message<?>` cannot be null in `postReceive()`.

Polishing - PR Comments

* Polishing code style
* Fix `PollableJmsChannel`
* Increase receive timeout in the `PollableJmsChannel`
This commit is contained in:
Gary Russell
2018-08-24 13:54:36 -04:00
committed by Artem Bilan
parent 4b9eaebd42
commit e6489abb52
7 changed files with 107 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
View File

@@ -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;

6
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
View File

@@ -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;

8
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java
View File

@@ -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<String>("test"));
Message<?> result = channel.receive(0);
assertNotNull(result);
assertEquals(2, invokedCount.get());
assertEquals(1, messageCount.get());
}

71
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java
View File

@@ -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<Object, Object> param = Collections.<Object, Object>singletonMap(1, 1);
Map<Object, Object> param = Collections.singletonMap(1, 1);
Object result = this.int2634Gateway.test2(param);
assertEquals(param, result);
@@ -405,7 +415,7 @@ public class GatewayInterfaceTests {
ListenableFuture<Thread> result2 = this.execGateway.test2(Thread.currentThread());
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<Thread> thread = new AtomicReference<Thread>();
final AtomicReference<Thread> thread = new AtomicReference<>();
result2.addCallback(new ListenableFutureCallback<Thread>() {
@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>(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

40
spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java
View File

@@ -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();

27
src/reference/asciidoc/channel.adoc
View File

@@ -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 <<channel-wiretap>>.
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`

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -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`