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 7c18432b23..293b952199 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 @@ -16,6 +16,11 @@ package org.springframework.integration.adapter; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; @@ -27,7 +32,9 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class AbstractSourceAdapter implements SourceAdapter { +public class AbstractSourceAdapter implements SourceAdapter, InitializingBean { + + protected Log logger = LogFactory.getLog(this.getClass()); private MessageChannel channel; @@ -54,8 +61,33 @@ public class AbstractSourceAdapter implements SourceAdapter { return this.mapper; } + public final void afterPropertiesSet() { + if (this.channel == null) { + throw new MessagingConfigurationException("'channel' is required"); + } + this.initialize(); + } + + /** + * Subclasses may implement this to take advantage of the initialization callback. + */ + protected void initialize() { + } + protected boolean sendToChannel(T object) { - Message message = this.mapper.toMessage(object); + Message message = null; + if (object instanceof Message) { + message = (Message) object; + } + else { + message = this.mapper.toMessage(object); + } + if (message == null) { + if (logger.isWarnEnabled()) { + logger.warn("unable to create Message from source object: " + object); + } + return false; + } if (this.sendTimeout < 0) { return this.channel.send(message); } 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 new file mode 100644 index 0000000000..ab105b9b99 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java @@ -0,0 +1,145 @@ +/* + * 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.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageListener; + +import org.springframework.beans.factory.InitializingBean; +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; +import org.springframework.jms.listener.AbstractJmsListeningContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.jms.support.converter.SimpleMessageConverter; +import org.springframework.util.Assert; + +/** + * A message-driven adapter for receiving JMS messages and sending to a channel. + * + * @author Mark Fisher + */ +public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter implements MessageListener, Lifecycle, + InitializingBean { + + private AbstractJmsListeningContainer container; + + private ConnectionFactory connectionFactory; + + private Destination destination; + + private String destinationName; + + private MessageConverter messageConverter = new SimpleMessageConverter(); + + private TaskExecutor taskExecutor; + + private ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); + + + public void setContainer(AbstractJmsListeningContainer container) { + this.container = container; + } + + public void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + } + + public void setDestination(Destination destination) { + this.destination = destination; + } + + public void setDestinationName(String destinationName) { + this.destinationName = destinationName; + } + + public void setMessageConverter(MessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' must not be null"); + this.messageConverter = messageConverter; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setPolicy(ConsumerPolicy policy) { + Assert.notNull(policy, "'policy' must not be null"); + this.policy = policy; + } + + @Override + public void initialize() { + if (this.container == null) { + initDefaultContainer(); + } + } + + private void initDefaultContainer() { + if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) { + throw new MessagingConfigurationException("If a 'container' reference is not provided, then " + + "'connectionFactory' and 'destination' (or 'destinationName') are required."); + } + DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer(); + dmlc.setConnectionFactory(this.connectionFactory); + if (this.destination != null) { + dmlc.setDestination(this.destination); + } + if (this.destinationName != null) { + dmlc.setDestinationName(this.destinationName); + } + dmlc.setReceiveTimeout(this.policy.getReceiveTimeout()); + dmlc.setConcurrentConsumers(this.policy.getConcurrency()); + dmlc.setMaxConcurrentConsumers(this.policy.getMaxConcurrency()); + dmlc.setMaxMessagesPerTask(this.policy.getMaxMessagesPerTask()); + dmlc.setAutoStartup(false); + dmlc.setMessageListener(this); + if (this.taskExecutor != null) { + dmlc.setTaskExecutor(this.taskExecutor); + } + dmlc.afterPropertiesSet(); + this.container = dmlc; + } + + public boolean isRunning() { + return container.isRunning(); + } + + public void start() { + container.start(); + } + + public void stop() { + container.stop(); + } + + public void onMessage(Message message) { + try { + this.sendToChannel(messageConverter.fromMessage(message)); + } + catch (JMSException e) { + throw new MessageHandlingException("failed to convert JMS Message", e); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java index 3b8e782fc1..547426b371 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java @@ -75,6 +75,17 @@ public class ConsumerPolicy { return policy; } + /** + * Factory method for a basic event-driven policy. + */ + public static ConsumerPolicy newEventDrivenPolicy() { + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setPeriod(-1); + policy.setConcurrency(1); + policy.setMaxConcurrency(1); + return policy; + } + public int getInitialDelay() { return this.initialDelay; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 68ec43bb56..1db01c914d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -58,6 +58,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private Map> targetAdapters = new ConcurrentHashMap>(); + private Map lifecycleComponents = new ConcurrentHashMap(); + private List dispatcherTasks = new CopyOnWriteArrayList(); private Map, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap, MessageReceivingExecutor>(); @@ -183,10 +185,19 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif ConsumerPolicy policy = dispatcher.getConsumerPolicy(); DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); this.addDispatcherTask(dispatcherTask); - if (logger.isInfoEnabled()) { - logger.info("registered source adapter '" + name + "'"); + } + if (adapter instanceof Lifecycle) { + this.lifecycleComponents.put(name, (Lifecycle) adapter); + if (this.isRunning()) { + ((Lifecycle) adapter).start(); + if (logger.isInfoEnabled()) { + logger.info("started source adapter '" + name + "'"); + } } } + if (logger.isInfoEnabled()) { + logger.info("registered source adapter '" + name + "'"); + } } public void registerTargetAdapter(String name, TargetAdapter targetAdapter) { @@ -231,7 +242,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif "' for endpoint '" + endpointName + "'"); } MessageReceivingExecutor executor = new MessageReceivingExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency()); - receiverExecutors.put(endpoint, executor); + this.receiverExecutors.put(endpoint, executor); + this.lifecycleComponents.put(endpointName + "-executor", executor); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); dispatcher.addExecutor(executor); @@ -302,8 +314,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (!this.isRunning()) { this.running = true; - for (MessageReceivingExecutor executor : receiverExecutors.values()) { - executor.start(); + for (Map.Entry entry : this.lifecycleComponents.entrySet()) { + entry.getValue().start(); + if (logger.isInfoEnabled()) { + logger.info("started lifecycle component '" + entry.getKey() + "'"); + } } for (DispatcherTask task : this.dispatcherTasks) { scheduleDispatcherTask(task); @@ -316,8 +331,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (this.isRunning()) { this.running = false; - for (MessageReceivingExecutor executor : receiverExecutors.values()) { - executor.stop(); + for (Map.Entry entry : this.lifecycleComponents.entrySet()) { + entry.getValue().stop(); + if (logger.isInfoEnabled()) { + logger.info("stopped lifecycle component '" + entry.getKey() + "'"); + } } this.dispatcherExecutor.shutdownNow(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java index e5a721f310..fad085015d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java @@ -57,9 +57,8 @@ public class UnicastMessageDispatcher extends AbstractMessageDispatcher { try { if (executor == null || !executor.isRunning()) { if (logger.isInfoEnabled()) { - logger.info("removing inactive executor"); + logger.info("skipping inactive executor"); } - iter.remove(); continue; } executor.processMessage(message);