Added 'defaultReplyDestination' to ChannelPublishingJmsMessageListener, and now throwing an Exception if there is a reply Message to send and neither a reply-to property or 'defaultReplyDestination' is available (INT-560 and INT-580).

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);
 			}
 		}

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");
+		}
+	}
+
+}