diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index e497f24939..061180376a 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -16,6 +16,7 @@ package org.springframework.integration.jms; +import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; @@ -24,6 +25,7 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessagingException; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.jms.listener.SessionAwareMessageListener; @@ -47,6 +49,8 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL private volatile boolean extractReplyPayload = true; + private volatile Destination defaultReplyDestination; + private volatile JmsHeaderMapper headerMapper; private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); @@ -84,6 +88,14 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL this.channelTemplate.setReceiveTimeout(replyTimeout); } + /** + * Specify the default reply Destination. If a request Message does not provide + * a 'JMSReplyTo' property, replies will be sent to this by default. + */ + public void setDefaultReplyDestination(Destination defaultReplyDestination) { + this.defaultReplyDestination = defaultReplyDestination; + } + /** * Provide a {@link MessageConverter} implementation to use when * converting between JMS Messages and Spring Integration Messages. @@ -152,11 +164,19 @@ public class ChannelPublishingJmsMessageListener implements SessionAwareMessageL else { Message replyMessage = this.channelTemplate.sendAndReceive(requestMessage); if (replyMessage != null) { + Destination destination = jmsMessage.getJMSReplyTo(); + if (destination == null) { + destination = this.defaultReplyDestination; + } + if (destination == null) { + throw new MessagingException(replyMessage, "Unable to send JMS reply. The request Message " + + "has no 'JMSReplyTo' property, and this listener has no 'defaultReplyDestination'."); + } javax.jms.Message jmsReply = this.messageConverter.toMessage(replyMessage, session); if (jmsReply.getJMSCorrelationID() == null) { jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID()); } - MessageProducer producer = session.createProducer(jmsMessage.getJMSReplyTo()); + MessageProducer producer = session.createProducer(destination); producer.send(jmsReply); } } diff --git a/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java new file mode 100644 index 0000000000..d395d2d375 --- /dev/null +++ b/org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListenerTests.java @@ -0,0 +1,78 @@ +/* + * 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; + +import javax.jms.JMSException; +import javax.jms.Session; + +import org.junit.Test; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.message.StringMessage; +import org.springframework.jms.support.converter.MessageConversionException; +import org.springframework.jms.support.converter.MessageConverter; + +/** + * @author Mark Fisher + */ +public class ChannelPublishingJmsMessageListenerTests { + + private final Session session = new StubSession("test"); + + + @Test(expected = MessagingException.class) + public void noReplyToAndNoDefault() throws JMSException { + final QueueChannel requestChannel = new QueueChannel(); + this.startBackgroundReplier(requestChannel); + ChannelPublishingJmsMessageListener listener = new ChannelPublishingJmsMessageListener(); + listener.setExpectReply(true); + listener.setRequestChannel(requestChannel); + listener.setMessageConverter(new TestMessageConverter()); + javax.jms.Message jmsMessage = session.createTextMessage("test"); + listener.afterPropertiesSet(); + listener.onMessage(jmsMessage, session); + } + + + private void startBackgroundReplier(final PollableChannel channel) { + new SimpleAsyncTaskExecutor().execute(new Runnable() { + public void run() { + Message request = channel.receive(5000); + Message reply = new StringMessage(((String) request.getPayload()).toUpperCase()); + ((MessageChannel) request.getHeaders().getReplyChannel()).send(reply, 5000); + } + }); + } + + + private static class TestMessageConverter implements MessageConverter { + + public Object fromMessage(javax.jms.Message message) throws JMSException, MessageConversionException { + return "test-from"; + } + + public javax.jms.Message toMessage(Object object, Session session) throws JMSException, MessageConversionException { + return new StubTextMessage("test-to"); + } + } + +}