JmsMessageDrivenSourceAdapter now delegates to ChannelPublishingJmsListener. The latter may also be used on its own.

spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java

@@ -53,6 +53,10 @@ public abstract class AbstractSourceAdapter<T> implements SourceAdapter, Initial
 		this.channel = channel;
 	}
 
+	protected MessageChannel getChannel() {
+		return this.channel;
+	}
+
 	public void setSendTimeout(long sendTimeout) {
 		this.sendTimeout = sendTimeout;
 	}

spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java

@@ -0,0 +1,96 @@
+/*
+ * 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.MessageListener;
+
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageMapper;
+import org.springframework.integration.message.SimplePayloadMessageMapper;
+import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.converter.SimpleMessageConverter;
+import org.springframework.util.Assert;
+
+/**
+ * JMS {@link MessageListener} implementation that converts the received JMS
+ * message into a Spring Integration message and then sends that to a channel.
+ * 
+ * @author Mark Fisher
+ */
+public class ChannelPublishingJmsListener implements MessageListener {
+
+	private MessageChannel channel;
+
+	private long timeout = -1;
+
+	private MessageConverter converter = new SimpleMessageConverter();
+
+	private MessageMapper mapper = new SimplePayloadMessageMapper();
+
+
+	public ChannelPublishingJmsListener() {
+	}
+
+	public ChannelPublishingJmsListener(MessageChannel channel) {
+		Assert.notNull(channel, "'channel' must not be null");
+		this.channel = channel;
+	}
+
+
+	public void setChannel(MessageChannel channel) {
+		Assert.notNull(channel, "'channel' must not be null");
+		this.channel = channel;
+	}
+
+	public void setTimeout(long timeout) {
+		this.timeout = timeout;
+	}
+
+	public void setMessageConverter(MessageConverter messageConverter) {
+		Assert.notNull(messageConverter, "'messageConverter' must not be null");
+		this.converter = messageConverter;
+	}
+
+	public void setMessageMapper(MessageMapper messageMapper) {
+		Assert.notNull(messageMapper, "'messageMapper' must not be null");
+		this.mapper = messageMapper;
+	}
+
+	public void onMessage(javax.jms.Message jmsMessage) {
+		if (this.channel == null) {
+			throw new MessagingConfigurationException("'channel' must not be null");
+		}
+		try {
+			Object payload = converter.fromMessage(jmsMessage);
+			Message messageToSend = mapper.toMessage(payload);
+			if (this.timeout < 0) {
+				this.channel.send(messageToSend);
+			}
+			else {
+				this.channel.send(messageToSend, timeout);
+			}
+		}
+		catch (JMSException e) {
+			throw new MessageDeliveryException("failed to convert JMS Message", e);
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java

@@ -18,14 +18,10 @@ package org.springframework.integration.adapter.jms;
 
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
 import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessagingConfigurationException;
 import org.springframework.integration.adapter.AbstractSourceAdapter;
 import org.springframework.integration.bus.ConsumerPolicy;
@@ -40,8 +36,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object> implements MessageListener, Lifecycle,
-		DisposableBean {
+public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object> implements Lifecycle, DisposableBean {
 
 	private AbstractJmsListeningContainer container;
 
@@ -57,6 +52,8 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
 
 	private ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
 
+	private long sendTimeout = -1;
+
 
 	public void setContainer(AbstractJmsListeningContainer container) {
 		this.container = container;
@@ -79,6 +76,10 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
 		this.messageConverter = messageConverter;
 	}
 
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
 	public void setTaskExecutor(TaskExecutor taskExecutor) {
 		this.taskExecutor = taskExecutor;
 	}
@@ -108,7 +109,11 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
 		dmlc.setMaxConcurrentConsumers(this.policy.getMaxConcurrency());
 		dmlc.setMaxMessagesPerTask(this.policy.getMaxMessagesPerTask());
 		dmlc.setAutoStartup(false);
-		dmlc.setMessageListener(this);
+		ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel());
+		listener.setMessageConverter(this.messageConverter);
+		listener.setMessageMapper(this.getMessageMapper());
+		listener.setTimeout(this.sendTimeout);
+		dmlc.setMessageListener(listener);
 		if (this.taskExecutor != null) {
 			dmlc.setTaskExecutor(this.taskExecutor);
 		}
@@ -132,13 +137,4 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
 		container.destroy();
 	}
 
-	public void onMessage(Message message) {
-		try {
-			this.sendToChannel(messageConverter.fromMessage(message));
-		}
-		catch (JMSException e) {
-			throw new MessageHandlingException("failed to convert JMS Message", e);
-		}
-	}
-
 }