Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added JmsMessageDrivenSourceAdapter.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-04 21:37:30 +00:00
parent 289e5ccc13
commit a9f7b7dbdf
5 changed files with 216 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<T> implements SourceAdapter {
public class AbstractSourceAdapter<T> implements SourceAdapter, InitializingBean {
protected Log logger = LogFactory.getLog(this.getClass());
private MessageChannel channel;
@@ -54,8 +61,33 @@ public class AbstractSourceAdapter<T> 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);
}

145
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java Normal file
View File

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

11
spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java
View File

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

32
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -58,6 +58,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
private Map<String, TargetAdapter<?>> targetAdapters = new ConcurrentHashMap<String, TargetAdapter<?>>();
private Map<String, Lifecycle> lifecycleComponents = new ConcurrentHashMap<String, Lifecycle>();
private List<DispatcherTask> dispatcherTasks = new CopyOnWriteArrayList<DispatcherTask>();
private Map<MessageReceiver<?>, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap<MessageReceiver<?>, 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<String, Lifecycle> 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<String, Lifecycle> entry : this.lifecycleComponents.entrySet()) {
entry.getValue().stop();
if (logger.isInfoEnabled()) {
logger.info("stopped lifecycle component '" + entry.getKey() + "'");
}
}
this.dispatcherExecutor.shutdownNow();
}

3
spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java
View File

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