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

AbstractMessageHandler now extends IntegrationObjectSupport

Browse Source
This commit is contained in:
Mark Fisher
2010-03-11 20:00:48 +00:00
parent 53e066bc1d
commit d0ab4dafb2
11 changed files with 50 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
View File

@@ -27,11 +27,9 @@ import java.util.concurrent.ScheduledFuture;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
@@ -108,8 +106,6 @@ public abstract class AbstractMessageBarrierHandler<T extends Collection<? exten
private volatile boolean autoStartup = true;
private volatile TaskScheduler taskScheduler;
private volatile ScheduledFuture<?> reaperFutureTask;
private volatile boolean initialized;
@@ -173,25 +169,19 @@ public abstract class AbstractMessageBarrierHandler<T extends Collection<? exten
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
Assert.notNull(taskScheduler, "taskScheduler must not be null");
this.taskScheduler = taskScheduler;
super.setTaskScheduler(taskScheduler);
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setBeanFactory(BeanFactory beanFactory) {
if (this.taskScheduler == null) {
this.taskScheduler = IntegrationContextUtils.getRequiredTaskScheduler(beanFactory);
}
}
public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
this.correlationStrategy = correlationStrategy;
}
public final void afterPropertiesSet() {
@Override
public final void onInit() {
synchronized (this.lifecycleMonitor) {
if (!this.initialized) {
if (this.trackedCorrelationIdCapacity > 0) {
@@ -216,8 +206,8 @@ public abstract class AbstractMessageBarrierHandler<T extends Collection<? exten
if (this.isRunning()) {
return;
}
Assert.state(this.taskScheduler != null, "TaskScheduler must not be null");
this.reaperFutureTask = this.taskScheduler.scheduleWithFixedDelay(
Assert.state(this.getTaskScheduler() != null, "TaskScheduler must not be null");
this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(
new PrunerTask(), this.reaperInterval);
}
}

16
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -18,13 +18,12 @@ package org.springframework.integration.aggregator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
@@ -82,7 +81,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler
private final IdTracker tracker = new IdTracker();
private final BlockingQueue<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
private volatile TaskScheduler taskScheduler;
private volatile ScheduledFuture<?> reaperFutureTask;
private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
private volatile long timeout = DEFAULT_TIMEOUT;
@@ -130,7 +128,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
super.setTaskScheduler(taskScheduler);
}
public void setTimeout(long timeout) {
@@ -162,11 +160,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler
this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
}
public void setBeanFactory(BeanFactory beanFactory) {
if (this.taskScheduler == null) {
this.taskScheduler = IntegrationContextUtils.getRequiredTaskScheduler(beanFactory);
}
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
@@ -215,6 +208,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler
};
}
@SuppressWarnings("unchecked")
private void store(Message<?> message, Object correlationKey) {
Message toStore = message;
if (!correlationKey.equals(message.getHeaders().getCorrelationId())) {
@@ -238,8 +232,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler
if (this.isRunning()) {
return;
}
Assert.state(this.taskScheduler != null, "'taskScheduler' must not be null");
this.reaperFutureTask = this.taskScheduler.scheduleWithFixedDelay(
Assert.state(this.getTaskScheduler() != null, "'taskScheduler' must not be null");
this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(
new PrunerTask(), this.reaperInterval);
}
}

3
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -22,6 +22,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.core.Ordered;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
@@ -37,7 +38,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class AbstractMessageHandler implements MessageHandler, Ordered {
public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, Ordered {
protected final Log logger = LogFactory.getLog(this.getClass());

17
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -16,9 +16,6 @@
package org.springframework.integration.handler;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
@@ -37,15 +34,13 @@ import org.springframework.util.Assert;
* @author Iwein Fuld
*/
public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler
implements MessageProducer, BeanFactoryAware {
implements MessageProducer {
public static final long DEFAULT_SEND_TIMEOUT = 1000;
private MessageChannel outputChannel;
private volatile ChannelResolver channelResolver;
private volatile boolean requiresReply = false;
private final MessageChannelTemplate channelTemplate;
@@ -77,7 +72,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
*/
public void setChannelResolver(ChannelResolver channelResolver) {
Assert.notNull(channelResolver, "'channelResolver' must not be null");
this.channelResolver = channelResolver;
super.setChannelResolver(channelResolver);
}
/**
@@ -88,12 +83,6 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
this.requiresReply = requiresReply;
}
public void setBeanFactory(BeanFactory beanFactory) {
if (this.channelResolver == null) {
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
}
/**
* {@inheritDoc}
*/
@@ -110,7 +99,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
}
return;
}
MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.channelResolver);
MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.getChannelResolver());
MessageHeaders requestHeaders = message.getHeaders();
this.handleResult(result, requestHeaders, replyChannel);
}

49
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractChannelNameResolvingMessageRouter.java
View File

@@ -21,15 +21,10 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.ConversionServiceFactory;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
@@ -42,24 +37,17 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Jonas Partner
*/
public abstract class AbstractChannelNameResolvingMessageRouter extends AbstractMessageRouter
implements BeanFactoryAware, InitializingBean {
private volatile ChannelResolver channelResolver;
private volatile ConversionService conversionService;
public abstract class AbstractChannelNameResolvingMessageRouter extends AbstractMessageRouter {
private volatile String prefix;
private volatile String suffix;
private volatile BeanFactory beanFactory;
private volatile boolean ignoreChannelNameResolutionFailures;
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
super.setChannelResolver(channelResolver);
}
public void setPrefix(String prefix) {
@@ -70,29 +58,23 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract
this.suffix = suffix;
}
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
public void setIgnoreChannelNameResolutionFailures(boolean ignoreChannelNameResolutionFailures) {
this.ignoreChannelNameResolutionFailures = ignoreChannelNameResolutionFailures;
}
public void afterPropertiesSet() {
if (this.channelResolver == null) {
Assert.notNull(beanFactory,
"either a ChannelResolver or BeanFactory is required");
this.channelResolver = new BeanFactoryChannelResolver(this.beanFactory);
}
@Override
public final void onInit() {
Assert.notNull(this.getChannelResolver(),
"either a ChannelResolver or BeanFactory is required");
}
protected MessageChannel resolveChannelForName(String channelName, Message<?> message) {
Assert.state(this.channelResolver != null,
Assert.state(this.getChannelResolver() != null,
"unable to resolve channel names, no ChannelResolver available");
MessageChannel channel = null;
try {
channel = this.channelResolver.resolveChannelName(channelName);
channel = this.getChannelResolver().resolveChannelName(channelName);
}
catch (ChannelResolutionException e) {
if (!ignoreChannelNameResolutionFailures)
@@ -141,7 +123,7 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract
else if (channelIndicator instanceof Collection) {
addToCollection(channels, (Collection<?>) channelIndicator, message);
}
else if (this.getConversionService().canConvert(channelIndicator.getClass(), String.class)) {
else if (this.getRequiredConversionService().canConvert(channelIndicator.getClass(), String.class)) {
addChannelFromString(channels,
this.getConversionService().convert(channelIndicator, String.class), message);
}
@@ -165,16 +147,11 @@ public abstract class AbstractChannelNameResolvingMessageRouter extends Abstract
}
}
private ConversionService getConversionService() {
if (this.conversionService == null) {
if (this.beanFactory != null) {
this.conversionService = IntegrationContextUtils.getConversionService(this.beanFactory);
}
if (this.conversionService == null) {
this.conversionService = ConversionServiceFactory.createDefaultConversionService();
}
private ConversionService getRequiredConversionService() {
if (this.getConversionService() == null) {
this.setConversionService(ConversionServiceFactory.createDefaultConversionService());
}
return this.conversionService;
return this.getConversionService();
}
/**

3
org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -90,7 +90,8 @@ public class RecipientListRouter extends AbstractMessageRouter implements Initia
this.channelMap = channelMap;
}
public void afterPropertiesSet() {
@Override
public final void onInit() {
Assert.notEmpty(this.channelMap, "a non-empty channel map is required");
}

9
org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -26,6 +26,7 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.support.ComponentMetadata;
/**
* Base class for Message-splitting handlers.
@@ -34,9 +35,6 @@ import org.springframework.integration.message.MessageBuilder;
*/
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
public static final String COMPONENT_TYPE_LABEL = "splitter";
@Override
@SuppressWarnings("unchecked")
protected final Object handleRequestMessage(Message<?> message) {
@@ -93,6 +91,11 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
}
}
@Override
protected void populateComponentMetadata(ComponentMetadata metadata) {
metadata.setComponentType("splitter");
}
/**
* Subclasses must override this method to split the received Message. The
* return value may be a Collection or Array. The individual elements may