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 bb0bb563aa..4eb8917233 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 @@ -23,6 +23,7 @@ import org.springframework.jms.core.JmsTemplate; /** * @author Mark Fisher + * @author Oleg Zhurakousky * @since 2.0 */ public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel { @@ -33,14 +34,22 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh public Message receive() { + if (!this.getInterceptors().preReceive(this)) { + return null; + } Object object = this.getJmsTemplate().receiveAndConvert(); + if (object == null) { return null; } + Message replyMessage = null; if (object instanceof Message) { - return (Message) object; + replyMessage = (Message) object; } - return MessageBuilder.withPayload(object).build(); + else { + replyMessage = MessageBuilder.withPayload(object).build(); + } + return this.getInterceptors().postReceive(replyMessage, this) ; } public Message receive(long timeout) { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java index 9569f57d1e..5f0de96100 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java @@ -16,17 +16,27 @@ package org.springframework.integration.jms; +import static junit.framework.Assert.assertNull; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.util.ArrayList; +import java.util.List; import javax.jms.Destination; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; import org.junit.Test; +import org.mockito.Mockito; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.ChannelInterceptor; import org.springframework.integration.jms.config.ActiveMqTestUtils; import org.springframework.integration.jms.config.JmsChannelFactoryBean; import org.springframework.integration.message.GenericMessage; @@ -87,5 +97,76 @@ public class PollableJmsChannelTests { assertNotNull(result2); assertEquals("bar", result2.getPayload()); } + + @Test + public void queueNameWithFalsePreReceiveInterceptors() throws Exception { + ActiveMqTestUtils.prepare(); + this.connectionFactory = new ActiveMQConnectionFactory(); + this.connectionFactory.setBrokerURL("vm://localhost"); + + JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false); + factoryBean.setConnectionFactory(this.connectionFactory); + factoryBean.setDestinationName("someDynamicQueue"); + factoryBean.setPubSubDomain(false); + List interceptorList = new ArrayList(); + ChannelInterceptor interceptor = spy(new SampleInterceptor(false)); + interceptorList.add(interceptor); + factoryBean.setInterceptors(interceptorList); + factoryBean.afterPropertiesSet(); + PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject(); + boolean sent1 = channel.send(new GenericMessage("foo")); + assertTrue(sent1); + Message result1 = channel.receive(10000); + assertNull(result1); + verify(interceptor, times(1)).preReceive(Mockito.any(MessageChannel.class)); + verify(interceptor, times(0)).postReceive(Mockito.any(Message.class), Mockito.any(MessageChannel.class)); + } + + @Test + public void queueNameWithTruePreReceiveInterceptors() throws Exception { + ActiveMqTestUtils.prepare(); + this.connectionFactory = new ActiveMQConnectionFactory(); + this.connectionFactory.setBrokerURL("vm://localhost"); + + JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false); + factoryBean.setConnectionFactory(this.connectionFactory); + factoryBean.setDestinationName("someDynamicQueue"); + factoryBean.setPubSubDomain(false); + List interceptorList = new ArrayList(); + ChannelInterceptor interceptor = spy(new SampleInterceptor(true)); + interceptorList.add(interceptor); + factoryBean.setInterceptors(interceptorList); + factoryBean.afterPropertiesSet(); + PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject(); + boolean sent1 = channel.send(new GenericMessage("foo")); + assertTrue(sent1); + Message result1 = channel.receive(10000); + assertNotNull(result1); + verify(interceptor, times(1)).preReceive(Mockito.any(MessageChannel.class)); + verify(interceptor, times(1)).postReceive(Mockito.any(Message.class), Mockito.any(MessageChannel.class)); + } + public static class SampleInterceptor implements ChannelInterceptor { + private final boolean preRecieveFlag; + public SampleInterceptor(boolean preRecieveFlag){ + this.preRecieveFlag = preRecieveFlag; + } + + public Message preSend(Message message, MessageChannel channel) { + return message; + } + + public void postSend(Message message, MessageChannel channel, + boolean sent) { + } + + public boolean preReceive(MessageChannel channel) { + return this.preRecieveFlag; + } + + public Message postReceive(Message message, MessageChannel channel) { + return message; + } + + } }