diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessor.java new file mode 100644 index 0000000000..19151ca83e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessor.java @@ -0,0 +1,48 @@ +/* + * Copyright 2002-2007 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.adapter.jms; + +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; + +import org.springframework.integration.message.MessageHeader; + +/** + * A {@link JmsMessagePostProcessor} that passes attributes from a + * {@link MessageHeader} to a JMS message before it is sent to its destination. + * + * @author Mark Fisher + */ +public class DefaultJmsMessagePostProcessor implements JmsMessagePostProcessor { + + public void postProcessJmsMessage(Message jmsMessage, MessageHeader header) throws JMSException { + Object jmsCorrelationId = header.getAttribute(JmsTargetAdapter.JMS_CORRELATION_ID); + if (jmsCorrelationId != null && (jmsCorrelationId instanceof String)) { + jmsMessage.setJMSCorrelationID((String) jmsCorrelationId); + } + Object jmsReplyTo = header.getAttribute(JmsTargetAdapter.JMS_REPLY_TO); + if (jmsReplyTo != null && (jmsReplyTo instanceof Destination)) { + jmsMessage.setJMSReplyTo((Destination) jmsReplyTo); + } + Object jmsType = header.getAttribute(JmsTargetAdapter.JMS_TYPE); + if (jmsType != null && (jmsType instanceof String)) { + jmsMessage.setJMSType((String) jmsType); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessagePostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessagePostProcessor.java new file mode 100644 index 0000000000..2b98a66757 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessagePostProcessor.java @@ -0,0 +1,34 @@ +/* + * Copyright 2002-2007 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.adapter.jms; + +import javax.jms.JMSException; +import javax.jms.Message; + +import org.springframework.integration.message.MessageHeader; + +/** + * Strategy interface for post-processing a JMS Message before it is sent to its + * destination. + * + * @author Mark Fisher + */ +public interface JmsMessagePostProcessor { + + void postProcessJmsMessage(Message jmsMessage, MessageHeader header) throws JMSException; + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsTargetAdapter.java index 33ed733ef8..28badc5174 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsTargetAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsTargetAdapter.java @@ -18,25 +18,40 @@ package org.springframework.integration.adapter.jms; import javax.jms.ConnectionFactory; import javax.jms.Destination; +import javax.jms.JMSException; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.adapter.AbstractTargetAdapter; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.core.MessagePostProcessor; +import org.springframework.jms.support.converter.SimpleMessageConverter; /** * A target adapter for sending JMS Messages. * * @author Mark Fisher */ -public class JmsTargetAdapter extends AbstractTargetAdapter implements InitializingBean { +public class JmsTargetAdapter implements MessageHandler, InitializingBean { + + public static final String JMS_CORRELATION_ID = "JMSCorrelationID"; + + public static final String JMS_REPLY_TO = "JMSReplyTo"; + + public static final String JMS_TYPE = "JMSType"; + private ConnectionFactory connectionFactory; private Destination destination; + private String destinationName; + private JmsTemplate jmsTemplate; + private JmsMessagePostProcessor jmsMessagePostProcessor = new DefaultJmsMessagePostProcessor(); + public JmsTargetAdapter(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; @@ -48,6 +63,12 @@ public class JmsTargetAdapter extends AbstractTargetAdapter implements I this.initJmsTemplate(); } + public JmsTargetAdapter(ConnectionFactory connectionFactory, String destinationName) { + this.connectionFactory = connectionFactory; + this.destinationName = destinationName; + this.initJmsTemplate(); + } + /** * No-arg constructor provided for convenience when configuring with * setters. Note that the initialization callback will validate. @@ -64,30 +85,55 @@ public class JmsTargetAdapter extends AbstractTargetAdapter implements I this.destination = destination; } + public void setDestinationName(String destinationName) { + this.destinationName = destinationName; + } + public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } + public void setJmsMessagePostProcessor(JmsMessagePostProcessor jmsMessagePostProcessor) { + this.jmsMessagePostProcessor = jmsMessagePostProcessor; + } + public void afterPropertiesSet() { if (this.jmsTemplate == null) { - if (this.connectionFactory == null || this.destination == null) { + if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) { throw new MessagingConfigurationException("Either a 'jmsTemplate' or " + - "*both* 'connectionFactory' and 'destination' are required."); + "*both* 'connectionFactory' and 'destination' (or 'destination-name') are required."); } this.initJmsTemplate(); } + if (this.jmsTemplate.getMessageConverter() == null) { + this.jmsTemplate.setMessageConverter(new SimpleMessageConverter()); + } } private void initJmsTemplate() { this.jmsTemplate = new JmsTemplate(); this.jmsTemplate.setConnectionFactory(this.connectionFactory); - this.jmsTemplate.setDefaultDestination(this.destination); + if (this.destination != null) { + this.jmsTemplate.setDefaultDestination(this.destination); + } + else { + this.jmsTemplate.setDefaultDestinationName(this.destinationName); + } } - @Override - protected boolean sendToTarget(Object object) { - this.jmsTemplate.convertAndSend(object); - return true; + public final Message handle(final Message message) { + if (message == null) { + return null; + } + this.jmsTemplate.convertAndSend(message.getPayload(), new MessagePostProcessor() { + public javax.jms.Message postProcessMessage(javax.jms.Message jmsMessage) throws JMSException { + if (jmsMessagePostProcessor != null) { + jmsMessagePostProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + } + return jmsMessage; + } + }); + return null; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java index b2d4a3081e..502cf60ebb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java @@ -47,8 +47,12 @@ public class JmsTargetAdapterParser extends AbstractSingleBeanDefinitionParser { private static final String DESTINATION_ATTRIBUTE = "destination"; + private static final String DESTINATION_NAME_ATTRIBUTE = "destination-name"; + private static final String DESTINATION_PROPERTY = "destination"; + private static final String DESTINATION_NAME_PROPERTY = "destinationName"; + private static final String CHANNEL_ATTRIBUTE = "channel"; private static final String HANDLER_PROPERTY = "handler"; @@ -72,21 +76,27 @@ public class JmsTargetAdapterParser extends AbstractSingleBeanDefinitionParser { String jmsTemplate = element.getAttribute(JMS_TEMPLATE_ATTRIBUTE); String connectionFactory = element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE); String destination = element.getAttribute(DESTINATION_ATTRIBUTE); + String destinationName = element.getAttribute(DESTINATION_NAME_ATTRIBUTE); RootBeanDefinition adapterDef = new RootBeanDefinition(JmsTargetAdapter.class); if (StringUtils.hasText(jmsTemplate)) { - if (StringUtils.hasText(connectionFactory) || StringUtils.hasText(destination)) { - throw new BeanCreationException("when providing a 'jms-template' reference, neither " + - "'connection-factory' or 'destination' should be provided."); + if (StringUtils.hasText(connectionFactory) || StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { + throw new BeanCreationException("when providing a 'jms-template' reference, none of " + + "'connection-factory', 'destination', or 'destination-name' should be provided."); } adapterDef.getPropertyValues().addPropertyValue(JMS_TEMPLATE_PROPERTY, new RuntimeBeanReference(jmsTemplate)); } - else if (StringUtils.hasText(connectionFactory) && StringUtils.hasText(destination)) { + else if (StringUtils.hasText(connectionFactory) && (StringUtils.hasText(destination) ^ StringUtils.hasText(destinationName))) { adapterDef.getPropertyValues().addPropertyValue(CONNECTION_FACTORY_PROPERTY, new RuntimeBeanReference(connectionFactory)); - adapterDef.getPropertyValues().addPropertyValue(DESTINATION_PROPERTY, new RuntimeBeanReference(destination)); + if (StringUtils.hasText(destination)) { + adapterDef.getPropertyValues().addPropertyValue(DESTINATION_PROPERTY, new RuntimeBeanReference(destination)); + } + else { + adapterDef.getPropertyValues().addPropertyValue(DESTINATION_NAME_PROPERTY, destinationName); + } } else { throw new BeanCreationException("either a 'jms-template' reference or both " + - "'connection-factory' and 'destination' references must be provided."); + "'connection-factory' and 'destination' (or 'destination-name') references must be provided."); } String channel = element.getAttribute(CHANNEL_ATTRIBUTE); Subscription subscription = new Subscription(channel); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index ba5783bd99..cdc6e7c3c2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -221,6 +221,7 @@ + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessorTests.java new file mode 100644 index 0000000000..cc947bf0cb --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessorTests.java @@ -0,0 +1,102 @@ +/* + * Copyright 2002-2007 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.adapter.jms; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; + +import javax.jms.Destination; +import javax.jms.JMSException; + +import org.junit.Test; + +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class DefaultJmsMessagePostProcessorTests { + + @Test + public void testJmsReplyTo() throws JMSException { + StringMessage message = new StringMessage("test1"); + Destination replyTo = new Destination() {}; + message.getHeader().setAttribute(JmsTargetAdapter.JMS_REPLY_TO, replyTo); + DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor(); + javax.jms.Message jmsMessage = new StubTextMessage(); + postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + assertNotNull(jmsMessage.getJMSReplyTo()); + assertSame(replyTo, jmsMessage.getJMSReplyTo()); + } + + @Test + public void testJmsReplyToIgnoredIfIncorrectType() throws JMSException { + StringMessage message = new StringMessage("test1"); + message.getHeader().setAttribute(JmsTargetAdapter.JMS_REPLY_TO, "not-a-destination"); + DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor(); + javax.jms.Message jmsMessage = new StubTextMessage(); + postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + assertNull(jmsMessage.getJMSReplyTo()); + } + + @Test + public void testJmsCorrelationId() throws JMSException { + StringMessage message = new StringMessage("test1"); + String jmsCorrelationId = "ABC-123"; + message.getHeader().setAttribute(JmsTargetAdapter.JMS_CORRELATION_ID, jmsCorrelationId); + DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor(); + javax.jms.Message jmsMessage = new StubTextMessage(); + postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + assertNotNull(jmsMessage.getJMSCorrelationID()); + assertEquals(jmsCorrelationId, jmsMessage.getJMSCorrelationID()); + } + + @Test + public void testJmsCorrelationIdIgnoredIfIncorrectType() throws JMSException { + StringMessage message = new StringMessage("test1"); + message.getHeader().setAttribute(JmsTargetAdapter.JMS_CORRELATION_ID, new Integer(123)); + DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor(); + javax.jms.Message jmsMessage = new StubTextMessage(); + postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + assertNull(jmsMessage.getJMSCorrelationID()); + } + + @Test + public void testJmsType() throws JMSException { + StringMessage message = new StringMessage("test1"); + String jmsType = "testing"; + message.getHeader().setAttribute(JmsTargetAdapter.JMS_TYPE, jmsType); + DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor(); + javax.jms.Message jmsMessage = new StubTextMessage(); + postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + assertNotNull(jmsMessage.getJMSType()); + assertEquals(jmsType, jmsMessage.getJMSType()); + } + + @Test + public void testJmsTypeIgnoredIfIncorrectType() throws JMSException { + StringMessage message = new StringMessage("test1"); + message.getHeader().setAttribute(JmsTargetAdapter.JMS_TYPE, new Integer(123)); + DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor(); + javax.jms.Message jmsMessage = new StubTextMessage(); + postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader()); + assertNull(jmsMessage.getJMSType()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubTextMessage.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubTextMessage.java index 185b1a39f3..08322c33eb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubTextMessage.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubTextMessage.java @@ -10,6 +10,13 @@ public class StubTextMessage implements TextMessage { private String text; + private Destination replyTo; + + private String correlationID; + + private String type; + + public String getText() throws JMSException { return this.text; } @@ -48,7 +55,7 @@ public class StubTextMessage implements TextMessage { } public String getJMSCorrelationID() throws JMSException { - return null; + return this.correlationID; } public byte[] getJMSCorrelationIDAsBytes() throws JMSException { @@ -80,7 +87,7 @@ public class StubTextMessage implements TextMessage { } public Destination getJMSReplyTo() throws JMSException { - return null; + return this.replyTo; } public long getJMSTimestamp() throws JMSException { @@ -88,7 +95,7 @@ public class StubTextMessage implements TextMessage { } public String getJMSType() throws JMSException { - return null; + return this.type; } public long getLongProperty(String name) throws JMSException { @@ -131,6 +138,7 @@ public class StubTextMessage implements TextMessage { } public void setJMSCorrelationID(String correlationID) throws JMSException { + this.correlationID = correlationID; } public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException { @@ -155,12 +163,14 @@ public class StubTextMessage implements TextMessage { } public void setJMSReplyTo(Destination replyTo) throws JMSException { + this.replyTo = replyTo; } public void setJMSTimestamp(long timestamp) throws JMSException { } public void setJMSType(String type) throws JMSException { + this.type = type; } public void setLongProperty(String name, long value) throws JMSException { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParserTests.java index 218a00c31b..57ef368a85 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParserTests.java @@ -39,4 +39,14 @@ public class JmsTargetAdapterParserTests { assertEquals("testChannel", endpoint.getSubscription().getChannelName()); } + @Test + public void testTargetAdapterWithConnectionFactoryAndDestinationName() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "targetAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); + DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) context.getBean("adapter"); + assertEquals(JmsTargetAdapter.class, endpoint.getHandler().getClass()); + assertEquals("adapter", endpoint.getName()); + assertEquals("testChannel", endpoint.getSubscription().getChannelName()); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/targetAdapterWithConnectionFactoryAndDestinationName.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/targetAdapterWithConnectionFactoryAndDestinationName.xml new file mode 100644 index 0000000000..c50fb87fd0 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/targetAdapterWithConnectionFactoryAndDestinationName.xml @@ -0,0 +1,27 @@ + + + + + + + + + + + + + + + + + +