INT-2740 MessageSelector w/ Pollable JMS Channel

Previously, the 'selector' attribute was ignored when
a JMS Channel was pollable rather than message-driven.

Add support for MessageSelector to PollableJmsChannel.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -24,21 +24,33 @@ import org.springframework.jms.core.JmsTemplate;
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @since 2.0
  */
 public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel {
 
+	private volatile String messageSelector;
+
 	public PollableJmsChannel(JmsTemplate jmsTemplate) {
 		super(jmsTemplate);
 	}
 
+	public void setMessageSelector(String messageSelector) {
+		this.messageSelector = messageSelector;
+	}
 
 	public Message<?> receive() {
 		if (!this.getInterceptors().preReceive(this)) {
  			return null;
  		}
-		Object object = this.getJmsTemplate().receiveAndConvert();
-		
+		Object object;
+		if (this.messageSelector == null) {
+			object = this.getJmsTemplate().receiveAndConvert();
+		}
+		else {
+			object = this.getJmsTemplate().receiveSelectedAndConvert(this.messageSelector);
+		}
+
 		if (object == null) {
 			return null;
 		}

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java

@@ -341,7 +341,11 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		else {
 			Assert.isTrue(!Boolean.TRUE.equals(this.pubSubDomain),
 					"A JMS Topic-backed 'publish-subscribe-channel' must be message-driven.");
-			this.channel = new PollableJmsChannel(this.jmsTemplate);
+			PollableJmsChannel pollableJmschannel = new PollableJmsChannel(this.jmsTemplate);
+			if (this.messageSelector != null) {
+				pollableJmschannel.setMessageSelector(this.messageSelector);
+			}
+			this.channel = pollableJmschannel;
 		}
 		if (!CollectionUtils.isEmpty(this.interceptors)) {
 			this.channel.setInterceptors(this.interceptors);

spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java

@@ -32,6 +32,9 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Session;
+import javax.jms.TextMessage;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.command.ActiveMQQueue;
@@ -46,6 +49,7 @@ import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.jms.connection.CachingConnectionFactory;
 import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.core.MessageCreator;
 
 /**
  * @author Mark Fisher
@@ -202,6 +206,42 @@ public class PollableJmsChannelTests {
 		assertTrue(message.get().toString().contains("persistent = false"));
 	}
 
+	@Test
+	public void selector() throws Exception {
+		ActiveMqTestUtils.prepare();
+		this.connectionFactory = new ActiveMQConnectionFactory();
+		this.connectionFactory.setBrokerURL("vm://localhost");
+		this.queue = new ActiveMQQueue("pollableJmsChannelSelectorTestQueue");
+
+		JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(false);
+		factoryBean.setConnectionFactory(this.connectionFactory);
+		factoryBean.setDestination(this.queue);
+
+		factoryBean.setMessageSelector("baz='qux'");
+
+		factoryBean.afterPropertiesSet();
+		PollableJmsChannel channel = (PollableJmsChannel) factoryBean.getObject();
+		boolean sent1 = channel.send(new GenericMessage<String>("foo"));
+		assertTrue(sent1);
+		Message<?> result1 = channel.receive(100);
+		assertNull(result1);
+
+		JmsTemplate jmsTemplate = new JmsTemplate(this.connectionFactory);
+		jmsTemplate.setDefaultDestinationName("pollableJmsChannelSelectorTestQueue");
+		jmsTemplate.send(new MessageCreator() {
+
+			public javax.jms.Message createMessage(Session session) throws JMSException {
+				TextMessage message = session.createTextMessage("bar");
+				message.setStringProperty("baz", "qux");
+				return message;
+			}
+		});
+
+		Message<?> result2 = channel.receive(1000);
+		assertNotNull(result2);
+		assertEquals("bar", result2.getPayload());
+	}
+
 	public static class SampleInterceptor implements ChannelInterceptor {
 		private final boolean preReceiveFlag;
 		public SampleInterceptor(boolean preReceiveFlag){

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests-context.xml

@@ -47,6 +47,9 @@
 
 	<jms:channel id="pollableQueueNameChannel" queue-name="foo" message-driven="false"/>
 
+	<jms:channel id="pollableWithSelectorChannel" queue="testQueue" message-driven="false"
+		selector="foo='bar'" />
+
 	<bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
 		<property name="physicalName" value="test.queue"/>
 	</bean>

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java

@@ -94,6 +94,9 @@ public class JmsChannelParserTests {
 	@Autowired
 	private MessageChannel pollableQueueNameChannel;
 
+	@Autowired
+	private MessageChannel pollableWithSelectorChannel;
+
 	@Autowired
 	private Topic topic;
 
@@ -245,6 +248,16 @@ public class JmsChannelParserTests {
 		assertEquals("foo", jmsTemplate.getDefaultDestinationName());
 	}
 
+	@Test
+	public void selectorPollableChannel() {
+		assertEquals(PollableJmsChannel.class, pollableWithSelectorChannel.getClass());
+		PollableJmsChannel channel = (PollableJmsChannel) pollableWithSelectorChannel;
+		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
+		JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate");
+		assertEquals(queue, jmsTemplate.getDefaultDestination());
+		assertEquals("foo='bar'", accessor.getPropertyValue("messageSelector"));
+	}
+
 	@Test
 	public void withPlaceholders() {
 		DefaultMessageListenerContainer container = TestUtils.getPropertyValue(withPlaceholders, "container", DefaultMessageListenerContainer.class);