From bee854c42c64c44b483cf2c1040451ca77616260 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sat, 12 Jan 2008 02:02:02 +0000 Subject: [PATCH] JmsMessageDrivenSourceAdapter now delegates to ChannelPublishingJmsListener. The latter may also be used on its own. --- .../adapter/AbstractSourceAdapter.java | 4 + .../jms/ChannelPublishingJmsListener.java | 96 +++++++++++++++++++ .../jms/JmsMessageDrivenSourceAdapter.java | 28 +++--- 3 files changed, 112 insertions(+), 16 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java index 9f50397e46..617571f8fe 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -53,6 +53,10 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial this.channel = channel; } + protected MessageChannel getChannel() { + return this.channel; + } + public void setSendTimeout(long sendTimeout) { this.sendTimeout = sendTimeout; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java new file mode 100644 index 0000000000..9dab5306b0 --- /dev/null +++ b/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); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java index b96dc42dd3..78ee301a9c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java +++ b/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 implements MessageListener, Lifecycle, - DisposableBean { +public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter implements Lifecycle, DisposableBean { private AbstractJmsListeningContainer container; @@ -57,6 +52,8 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter 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 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 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 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); - } - } - }