From a3f029695dccf749c257f282f9e436dd24c5c404 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 4 Sep 2009 21:16:52 +0000 Subject: [PATCH] INT-764 The 'selector' attribute is now supported on the , , and elements within the JMS namespace. Any valid JMS Message selector expression may be provided. --- .../jms/JmsDestinationPollingSource.java | 13 +++++- .../JmsInboundChannelAdapterParser.java | 3 +- .../JmsMessageDrivenEndpointParser.java | 3 +- .../jms/config/spring-integration-jms-2.0.xsd | 7 +++ .../integration/jms/StubConsumer.java | 15 +++++-- .../integration/jms/StubSession.java | 6 +-- .../JmsInboundChannelAdapterParserTests.java | 12 ++++- .../config/JmsInboundGatewayParserTests.java | 15 ++++++- ...essageDrivenChannelAdapterParserTests.java | 45 +++++++++++++++++++ .../inboundGatewayWithMessageSelector.xml | 30 +++++++++++++ .../config/jmsInboundWithMessageSelector.xml | 45 +++++++++++++++++++ 11 files changed, 181 insertions(+), 13 deletions(-) create mode 100644 org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java create mode 100644 org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithMessageSelector.xml create mode 100644 org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithMessageSelector.xml diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java index c3ba65e68f..242795e416 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -37,6 +37,8 @@ public class JmsDestinationPollingSource extends AbstractJmsTemplateBasedAdapter private volatile boolean extractPayload = true; + private volatile String messageSelector; + public JmsDestinationPollingSource(JmsTemplate jmsTemplate) { super(jmsTemplate); @@ -51,6 +53,13 @@ public class JmsDestinationPollingSource extends AbstractJmsTemplateBasedAdapter } + /** + * Specify a JMS Message Selector expression to use when receiving Messages. + */ + public void setMessageSelector(String messageSelector) { + this.messageSelector = messageSelector; + } + /** * Specify whether the payload should be extracted from each received JMS * Message to be used as the Spring Integration Message payload. @@ -65,7 +74,7 @@ public class JmsDestinationPollingSource extends AbstractJmsTemplateBasedAdapter @SuppressWarnings("unchecked") public Message receive() { - Object receivedObject = this.getJmsTemplate().receiveAndConvert(); + Object receivedObject = this.getJmsTemplate().receiveSelectedAndConvert(this.messageSelector); if (receivedObject == null) { return null; } diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java index 82970d0ceb..e6c30c052d 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -78,6 +78,7 @@ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChanne if (StringUtils.hasText(headerMapper)) { builder.addPropertyReference(JmsAdapterParserUtils.HEADER_MAPPER_PROPERTY, headerMapper); } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver"); return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 9059ebdbe3..071cffa305 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -46,7 +46,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE, "destination-resolver", "transaction-manager", "pub-sub-domain", "concurrent-consumers", "max-concurrent-consumers", - "max-messages-per-task", "idle-task-execution-limit" + "max-messages-per-task", "idle-task-execution-limit", "selector" }; @@ -123,6 +123,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "destination-resolver"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "pub-sub-domain"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-concurrent-consumers"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-messages-per-task"); diff --git a/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd b/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd index 3024c6181b..13da511377 100644 --- a/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd +++ b/org.springframework.integration.jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd @@ -433,6 +433,13 @@ + + + + A JMS Message Selector expression. + + + diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubConsumer.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubConsumer.java index 4e69c78f26..32c85024ab 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubConsumer.java +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2009 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. @@ -28,9 +28,12 @@ public class StubConsumer implements MessageConsumer { private String messageText; + private String messageSelector; - public StubConsumer(String messageText) { + + public StubConsumer(String messageText, String messageSelector) { this.messageText = messageText; + this.messageSelector = messageSelector; } @@ -42,12 +45,16 @@ public class StubConsumer implements MessageConsumer { } public String getMessageSelector() throws JMSException { - return null; + return this.messageSelector; } public Message receive() throws JMSException { StubTextMessage message = new StubTextMessage(); - message.setText(this.messageText); + String text = this.messageText; + if (this.messageSelector != null) { + text += " [with selector: " + this.messageSelector + "]"; + } + message.setText(text); return message; } diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubSession.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubSession.java index b3995afe87..11211aab17 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubSession.java +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/StubSession.java @@ -69,16 +69,16 @@ public class StubSession implements Session { } public MessageConsumer createConsumer(Destination destination) throws JMSException { - return new StubConsumer(this.messageText); + return new StubConsumer(this.messageText, null); } public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException { - return new StubConsumer(this.messageText); + return new StubConsumer(this.messageText, messageSelector); } public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean NoLocal) throws JMSException { - return new StubConsumer(this.messageText); + return new StubConsumer(this.messageText, messageSelector); } public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException { diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParserTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParserTests.java index 60d3934800..4efa6ab23f 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParserTests.java +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -128,4 +128,14 @@ public class JmsInboundChannelAdapterParserTests { assertEquals(new Integer(123), message.getHeaders().get("testAttribute")); } + @Test + public void adapterWithMessageSelector() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "jmsInboundWithMessageSelector.xml", this.getClass()); + PollableChannel output = (PollableChannel) context.getBean("output1"); + Message message = output.receive(timeoutOnReceive); + assertNotNull("message should not be null", message); + assertEquals("test [with selector: TestProperty = 'foo']", message.getPayload()); + } + } diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java index cc1e852be8..b1d54eddf3 100644 --- a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -249,4 +249,17 @@ public class JmsInboundGatewayParserTests { gateway.stop(); } + @Test + public void testGatewayWithMessageSelector() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "inboundGatewayWithMessageSelector.xml", this.getClass()); + JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("gatewayWithMessageSelector"); + gateway.start(); + AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) + new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + String messageSelector = (String) new DirectFieldAccessor(container).getPropertyValue("messageSelector"); + assertEquals("TestProperty = 'foo'", messageSelector); + gateway.stop(); + } + } diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java new file mode 100644 index 0000000000..775f6eda2b --- /dev/null +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java @@ -0,0 +1,45 @@ +/* + * Copyright 2002-2009 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.jms.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.core.Message; + +/** + * @author Mark Fisher + */ +public class JmsMessageDrivenChannelAdapterParserTests { + + long timeoutOnReceive = 3000; + + @Test + public void adapterWithMessageSelector() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "jmsInboundWithMessageSelector.xml", this.getClass()); + PollableChannel output = (PollableChannel) context.getBean("output2"); + Message message = output.receive(timeoutOnReceive); + assertNotNull("message should not be null", message); + assertEquals("test [with selector: TestProperty = 'foo']", message.getPayload()); + } + +} diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithMessageSelector.xml b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithMessageSelector.xml new file mode 100644 index 0000000000..34e08f3542 --- /dev/null +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/inboundGatewayWithMessageSelector.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithMessageSelector.xml b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithMessageSelector.xml new file mode 100644 index 0000000000..bc1a9327ae --- /dev/null +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithMessageSelector.xml @@ -0,0 +1,45 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +