From 39247b5f3a3feb702f8d2a657dd9f38b22a22ce9 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 25 Nov 2008 22:51:31 +0000 Subject: [PATCH] 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). --- .../AbstractMessageBarrierHandler.java | 23 +++++++++++++++++-- .../AggregatorAnnotationPostProcessor.java | 1 + .../xml/AbstractConsumerEndpointParser.java | 8 ++----- .../config/xml/AggregatorParser.java | 1 + 4 files changed, 25 insertions(+), 8 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java index ef51c67462..1b66bf778c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java @@ -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>, 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> protected volatile BlockingQueue trackedCorrelationIds; + private volatile boolean autoStartup = true; + private volatile boolean initialized; private volatile TaskScheduler taskScheduler; @@ -107,11 +112,11 @@ public abstract class AbstractMessageBarrierHandler> 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> } 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(this.trackedCorrelationIdCapacity); + if (this.autoStartup) { + this.start(); + } this.initialized = true; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index 025ff2a640..81a86115b8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java index 47e8e9dd86..48b9838d9f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java @@ -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); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index 58cb747ee9..c128b03130 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -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);