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

AbstractMessageBarrierHandler is now BeanFactoryAware so that it can access the TaskScheduler. It also has an 'autoStartup' property with a default value of 'true' so that the pruner task will start upon inititialization (INT-495).

Browse Source
This commit is contained in:
Mark Fisher
2008-11-25 22:51:31 +00:00
parent cc7c053d87
commit 39247b5f3a
4 changed files with 25 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -28,8 +28,11 @@ import java.util.concurrent.TimeUnit;
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.handler.AbstractMessageHandler;
@@ -66,7 +69,7 @@ import org.springframework.util.Assert;
* @author Marius Bogoevici
*/
public abstract class AbstractMessageBarrierHandler<T extends Map<K, Message<?>>, K>
extends AbstractMessageHandler implements InitializingBean {
extends AbstractMessageHandler implements BeanFactoryAware, InitializingBean {
public final static long DEFAULT_SEND_TIMEOUT = 1000;
@@ -96,6 +99,8 @@ public abstract class AbstractMessageBarrierHandler<T extends Map<K, Message<?>>
protected volatile BlockingQueue<Object> trackedCorrelationIds;
private volatile boolean autoStartup = true;
private volatile boolean initialized;
private volatile TaskScheduler taskScheduler;
@@ -107,11 +112,11 @@ public abstract class AbstractMessageBarrierHandler<T extends Map<K, Message<?>>
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
/**
* Specify a channel for sending Messages that arrive after their
* aggregation group has either completed or timed-out.
@@ -158,11 +163,25 @@ public abstract class AbstractMessageBarrierHandler<T extends Map<K, Message<?>>
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
Assert.notNull(taskScheduler, "taskScheduler must not be null");
this.taskScheduler = 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 afterPropertiesSet() {
this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
if (this.autoStartup) {
this.start();
}
this.initialized = true;
}

1
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -63,6 +63,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
aggregator.setReaperInterval(annotation.reaperInterval());
aggregator.setTimeout(annotation.timeout());
aggregator.setTrackedCorrelationIdCapacity(annotation.trackedCorrelationIdCapacity());
aggregator.setBeanFactory(this.beanFactoryAccessor.getBeanFactory());
aggregator.afterPropertiesSet();
return aggregator;
}

8
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java
View File

@@ -40,10 +40,6 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
protected static final String METHOD_ATTRIBUTE = "method";
protected static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String POLLER_ELEMENT = "poller";
@Override
protected boolean shouldGenerateId() {
@@ -67,7 +63,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
@Override
protected final AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
BeanDefinitionBuilder handlerBuilder = this.parseHandler(element, parserContext);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, "output-channel");
AbstractBeanDefinition handlerBeanDefinition = handlerBuilder.getBeanDefinition();
if (!element.hasAttribute(this.getInputChannelAttributeName())) {
return handlerBeanDefinition;
@@ -84,7 +80,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
}
builder.addPropertyValue("inputChannelName", inputChannelName);
Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
if (pollerElement != null) {
IntegrationNamespaceUtils.configurePollerMetadata(pollerElement, builder, parserContext);
}

1
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -69,6 +69,7 @@ public class AggregatorParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
final String completionStrategyRef = element.getAttribute(COMPLETION_STRATEGY_REF_ATTRIBUTE);
final String completionStrategyMethod = element.getAttribute(COMPLETION_STRATEGY_METHOD_ATTRIBUTE);