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

Message Endpoints and the SimpleTaskScheduler now manage their own lifecycles. The ApplicationContextMessageBus is no longer necessary (part of INT-462). The MessagePublishingErrorHandler now detects the default error channel within the beanFactory if necessary (INT-464).

Browse Source
This commit is contained in:
Mark Fisher
2008-11-11 20:11:21 +00:00
parent 7c28053b0f
commit f4ccde6257
52 changed files with 759 additions and 1076 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
org.springframework.integration/src/main/java/org/springframework/integration/bus/ApplicationContextMessageBus.java
View File

@@ -16,175 +16,7 @@
package org.springframework.integration.bus;
import java.util.Collection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.TaskSchedulerAware;
import org.springframework.util.Assert;
/**
* Spring Integration's standard Message Bus implementation. Serves as a
* registry for Messages Endpoints. Manages their lifecycle, activates
* subscriptions, and schedules pollers by delegating to a
* {@link TaskScheduler}. Retrieves MessageChannels from the
* ApplicationContext based on bean name.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class ApplicationContextMessageBus implements ApplicationContextAware, ApplicationListener, Lifecycle, DisposableBean {
public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel";
private final Log logger = LogFactory.getLog(this.getClass());
private volatile TaskScheduler taskScheduler;
private volatile ApplicationContext applicationContext;
private volatile boolean autoStartup = true;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Assert.notNull(applicationContext, "'applicationContext' must not be null");
Assert.state(!(applicationContext.getBeanNamesForType(this.getClass()).length > 1),
"Only one instance of '" + this.getClass().getSimpleName() + "' is allowed per ApplicationContext.");
this.applicationContext = applicationContext;
}
/**
* Set the {@link TaskScheduler} to use for scheduling message dispatchers.
*/
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
/**
* Set whether to automatically start the bus after initialization.
* <p>Default is 'true'; set this to 'false' to allow for manual startup
* through the {@link #start()} method.
*/
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
private Collection<MessageEndpoint> getEndpoints() {
GenericBeanFactoryAccessor accessor = new GenericBeanFactoryAccessor(this.applicationContext);
return accessor.getBeansOfType(MessageEndpoint.class).values();
}
private void activateEndpoints() {
for (MessageEndpoint endpoint : this.getEndpoints()) {
if (endpoint != null) {
this.activateEndpoint(endpoint);
}
}
}
private void deactivateEndpoints() {
for (MessageEndpoint endpoint : this.getEndpoints()) {
if (endpoint != null) {
this.deactivateEndpoint(endpoint);
}
}
}
private void activateEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
if (endpoint instanceof TaskSchedulerAware) {
((TaskSchedulerAware) endpoint).setTaskScheduler(this.taskScheduler);
}
if (endpoint instanceof Lifecycle) {
((Lifecycle) endpoint).start();
}
if (logger.isInfoEnabled()) {
logger.info("activated endpoint '" + endpoint + "'");
}
}
private void deactivateEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
if (endpoint instanceof Lifecycle) {
((Lifecycle) endpoint).stop();
if (this.logger.isInfoEnabled()) {
logger.info("deactivated endpoint '" + endpoint + "'");
}
}
}
// Lifecycle implementation
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.running;
}
}
public void start() {
if (this.running) {
return;
}
Assert.notNull(this.applicationContext, "ApplicationContext must not be null");
Assert.notNull(this.taskScheduler, "TaskScheduler must not be null");
synchronized (this.lifecycleMonitor) {
this.activateEndpoints();
this.taskScheduler.start();
}
this.running = true;
this.applicationContext.publishEvent(new MessageBusStartedEvent(this));
if (logger.isInfoEnabled()) {
logger.info("message bus started");
}
}
public void stop() {
if (!this.isRunning()) {
return;
}
synchronized (this.lifecycleMonitor) {
this.deactivateEndpoints();
this.running = false;
this.taskScheduler.stop();
}
this.applicationContext.publishEvent(new MessageBusStoppedEvent(this));
if (logger.isInfoEnabled()) {
logger.info("message bus stopped");
}
}
// ApplicationListener implementation
public void onApplicationEvent(ApplicationEvent event) {
if (event instanceof ContextRefreshedEvent && this.autoStartup) {
this.start();
}
}
// DisposableBean implementation
public void destroy() throws Exception {
this.stop();
if (this.taskScheduler instanceof DisposableBean) {
((DisposableBean) this.taskScheduler).destroy();
}
}
// TODO: placeholder only, delete after handling all dependencies
public class ApplicationContextMessageBus {
}

5
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
View File

@@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
@@ -97,6 +98,10 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
}
private MessageChannel resolveErrorChannel(Message<?> failedMessage) {
if (this.defaultErrorChannel == null && this.channelResolver != null) {
this.defaultErrorChannel = this.channelResolver.resolveChannelName(
IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
}
if (failedMessage == null || failedMessage.getHeaders().getErrorChannel() == null) {
return this.defaultErrorChannel;
}

44
org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -22,11 +22,13 @@ import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageHandler;
@@ -39,7 +41,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, InitializingBean {
public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, InitializingBean, ApplicationListener {
private final MessageHandler handler;
@@ -61,7 +63,7 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
private volatile ConfigurableBeanFactory beanFactory;
private volatile MessageEndpoint endpoint;
private volatile AbstractEndpoint endpoint;
private volatile boolean initialized;
@@ -74,10 +76,6 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
}
public void setBeanName(String beanName) {
this.beanName = beanName;
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
@@ -106,14 +104,22 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
this.transactionDefinition = transactionDefinition;
}
public void setBeanName(String beanName) {
this.beanName = beanName;
}
public void setBeanFactory(BeanFactory beanFactory) {
Assert.isInstanceOf(ConfigurableBeanFactory.class, beanFactory,
"a ConfigurableBeanFactory is required");
this.beanFactory = (ConfigurableBeanFactory) beanFactory;
}
public void afterPropertiesSet() {
Assert.hasText(this.inputChannelName, "inputChannelName is required");
public void afterPropertiesSet() throws Exception {
this.initializeEndpoint();
}
public boolean isSingleton() {
return true;
}
public Object getObject() throws Exception {
@@ -125,20 +131,17 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
public Class<?> getObjectType() {
if (this.endpoint == null) {
return MessageEndpoint.class;
return AbstractEndpoint.class;
}
return endpoint.getClass();
return this.endpoint.getClass();
}
public boolean isSingleton() {
return true;
}
private void initializeEndpoint() {
private void initializeEndpoint() throws Exception {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
Assert.hasText(this.inputChannelName, "inputChannelName is required");
Assert.isTrue(this.beanFactory.containsBean(this.inputChannelName),
"no such input channel '" + this.inputChannelName + "' for endpoint '" + this.beanName + "'");
MessageChannel channel = (MessageChannel)
@@ -166,8 +169,17 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
throw new IllegalArgumentException(
"unsupported channel type: [" + channel.getClass() + "]");
}
this.endpoint.setBeanName(this.beanName);
this.endpoint.setBeanFactory(this.beanFactory);
if (this.endpoint instanceof InitializingBean) {
((InitializingBean) this.endpoint).afterPropertiesSet();
}
this.initialized = true;
}
}
public void onApplicationEvent(ApplicationEvent event) {
this.endpoint.onApplicationEvent(event);
}
}

10
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -20,7 +20,6 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
import org.springframework.core.annotation.AnnotationUtils;
@@ -64,15 +63,6 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
MessageEndpoint endpoint = this.createEndpoint(handler, annotation, pollerAnnotation);
if (endpoint != null) {
if (endpoint instanceof InitializingBean) {
try {
((InitializingBean) endpoint).afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(
"failed to initialize annotation-based consumer", e);
}
}
return endpoint;
}
return handler;

15
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -18,6 +18,9 @@ package org.springframework.integration.config.annotation;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.ChannelAdapter;
@@ -83,6 +86,18 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
String annotationName = ClassUtils.getShortNameAsProperty(annotation.annotationType());
String endpointName = beanName + "." + method.getName() + "." + annotationName;
this.beanFactory.registerSingleton(endpointName, endpoint);
// TODO: move this to IntegrationContextUtils?... common with MAPP
if (endpoint instanceof BeanFactoryAware) {
((BeanFactoryAware) endpoint).setBeanFactory(beanFactory);
}
if (endpoint instanceof InitializingBean) {
try {
((InitializingBean) endpoint).afterPropertiesSet();
}
catch (Exception e) {
throw new BeanInitializationException("failed to initialize annotated channel adapter", e);
}
}
}
return bean;
}

35
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -21,18 +21,22 @@ import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.ChannelAdapter;
@@ -40,7 +44,6 @@ import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
@@ -55,9 +58,7 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean {
private volatile Lifecycle messageBus;
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean, ApplicationListener {
private volatile ConfigurableListableBeanFactory beanFactory;
@@ -65,6 +66,8 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors =
new HashMap<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>>();
private Set<ApplicationListener> listeners = new HashSet<ApplicationListener>();
public void setBeanFactory(BeanFactory beanFactory) {
Assert.isAssignable(ConfigurableListableBeanFactory.class, beanFactory.getClass(),
@@ -74,7 +77,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
public void afterPropertiesSet() {
Assert.notNull(this.beanFactory, "BeanFactory must not be null");
this.messageBus = (Lifecycle) this.beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory));
postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.beanFactory));
postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.beanFactory));
@@ -108,8 +110,19 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
((BeanNameAware) result).setBeanName(endpointBeanName);
}
beanFactory.registerSingleton(endpointBeanName, result);
if (messageBus.isRunning() && result instanceof Lifecycle) {
((Lifecycle) result).start();
if (result instanceof BeanFactoryAware) {
((BeanFactoryAware) result).setBeanFactory(beanFactory);
}
if (result instanceof InitializingBean) {
try {
((InitializingBean) result).afterPropertiesSet();
}
catch (Exception e) {
throw new BeanInitializationException("failed to initialize annotated component", e);
}
}
if (result instanceof ApplicationListener) {
listeners.add((ApplicationListener) result);
}
}
}
@@ -158,4 +171,10 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
return name;
}
public void onApplicationEvent(ApplicationEvent event) {
for (ApplicationListener listener : listeners) {
listener.onApplicationEvent(event);
}
}
}

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

@@ -87,6 +87,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractSingleBeanD
@Override
protected final void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.registerTaskSchedulerIfNecessary(parserContext.getRegistry());
BeanDefinitionBuilder consumerBuilder = this.parseConsumer(element, parserContext);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerBuilder, element, OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerBuilder, element, SELECTOR_ATTRIBUTE);

5
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java
View File

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.util.LifecycleSupport.AutoStartMode;
import org.springframework.util.Assert;
import org.springframework.util.xml.DomUtils;
@@ -52,6 +53,10 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
else {
adapterBuilder.addPropertyValue("trigger", new IntervalTrigger(this.getDefaultPollInterval()));
}
String autoStart = element.getAttribute("auto-startup");
if ("false".equals(autoStart)) {
adapterBuilder.addPropertyValue("autoStartMode", AutoStartMode.NONE);
}
return adapterBuilder.getBeanDefinition();
}

36
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -23,11 +23,19 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.scheduling.CronTrigger;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.Assert;
@@ -195,4 +203,32 @@ public abstract class IntegrationNamespaceUtils {
targetBuilder.addPropertyValue("transactionDefinition", txDefinition);
}
/**
* Register a TaskScheduler in the given BeanDefinitionRegistry if not yet present.
* The bean name for which this is checking is defined by the constant
* {@link IntegrationContextUtils#TASK_SCHEDULER_BEAN_NAME}.
*/
public static synchronized void registerTaskSchedulerIfNecessary(BeanDefinitionRegistry registry) {
if (!registry.containsBeanDefinition(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
RootBeanDefinition errorChannelDef = new RootBeanDefinition(QueueChannel.class);
BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder(
errorChannelDef, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, registry);
}
TaskExecutor taskExecutor = null;
if (!registry.containsBeanDefinition(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)) {
taskExecutor = IntegrationContextUtils.createTaskExecutor(2, 100, 0, "integration-main-");
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(SimpleTaskScheduler.class);
schedulerBuilder.addConstructorArgValue(taskExecutor);
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MessagePublishingErrorHandler.class);
errorHandlerBuilder.addPropertyReference("defaultErrorChannel", IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
String errorHandlerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
errorHandlerBuilder.getBeanDefinition(), registry);
schedulerBuilder.addPropertyReference("errorHandler", errorHandlerBeanName);
BeanDefinitionHolder schedulerHolder = new BeanDefinitionHolder(
schedulerBuilder.getBeanDefinition(), IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME);
BeanDefinitionReaderUtils.registerBeanDefinition(schedulerHolder, registry);
}
}
}

77
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MessageBusParser.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.config.xml;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import org.w3c.dom.Element;
@@ -27,21 +26,14 @@ import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.context.event.SimpleApplicationEventMulticaster;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.bus.ApplicationContextMessageBus;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.integration.context.IntegrationContextUtils;
/**
* Parser for the &lt;message-bus&gt; element of the integration namespace.
@@ -51,35 +43,27 @@ import org.springframework.util.StringUtils;
*/
public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
public static final String MESSAGE_BUS_BEAN_NAME = "internal.MessageBus";
public static final String TASK_SCHEDULER_BEAN_NAME = "taskScheduler";
private static final String MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.MessagingAnnotationPostProcessor";
private static final String TASK_SCHEDULER_ATTRIBUTE = "task-scheduler";
private static final String ASYNC_EVENT_MULTICASTER_ATTRIBUTE = "configure-async-event-multicaster";
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
Assert.state(!parserContext.getRegistry().containsBeanDefinition(MESSAGE_BUS_BEAN_NAME),
"Only one Message Bus is allowed per ApplicationContext.");
return MESSAGE_BUS_BEAN_NAME;
}
@Override
protected Class<?> getBeanClass(Element element) {
return ApplicationContextMessageBus.class;
}
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
return "messageBus";
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !TASK_SCHEDULER_ATTRIBUTE.equals(attributeName) &&
!ASYNC_EVENT_MULTICASTER_ATTRIBUTE.equals(attributeName) &&
return !ASYNC_EVENT_MULTICASTER_ATTRIBUTE.equals(attributeName) &&
!"enable-annotations".equals(attributeName) &&
super.isEligibleAttribute(attributeName);
}
@@ -87,35 +71,9 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
super.doParse(element, parserContext, builder);
String taskSchedulerRef = element.getAttribute(TASK_SCHEDULER_ATTRIBUTE);
TaskExecutor taskExecutor= null;
if (!parserContext.getRegistry().containsBeanDefinition(ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME)) {
RootBeanDefinition errorChannelDef = new RootBeanDefinition(QueueChannel.class);
BeanDefinitionHolder errorChannelHolder = new BeanDefinitionHolder(
errorChannelDef, ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME);
BeanDefinitionReaderUtils.registerBeanDefinition(errorChannelHolder, parserContext.getRegistry());
}
if (StringUtils.hasText(taskSchedulerRef)) {
builder.addPropertyReference("taskScheduler", taskSchedulerRef);
}
else {
taskExecutor = this.createTaskExecutor(2, 100, 0, "message-bus-");
BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(SimpleTaskScheduler.class);
schedulerBuilder.addConstructorArgValue(taskExecutor);
BeanDefinitionBuilder errorHandlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MessagePublishingErrorHandler.class);
errorHandlerBuilder.addPropertyReference("defaultErrorChannel", ApplicationContextMessageBus.ERROR_CHANNEL_BEAN_NAME);
String errorHandlerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
errorHandlerBuilder.getBeanDefinition(), parserContext.getRegistry());
schedulerBuilder.addPropertyReference("errorHandler", errorHandlerBeanName);
BeanDefinitionHolder schedulerHolder = new BeanDefinitionHolder(
schedulerBuilder.getBeanDefinition(), TASK_SCHEDULER_BEAN_NAME);
BeanDefinitionReaderUtils.registerBeanDefinition(schedulerHolder, parserContext.getRegistry());
builder.addPropertyReference("taskScheduler", TASK_SCHEDULER_BEAN_NAME);
}
IntegrationNamespaceUtils.registerTaskSchedulerIfNecessary(parserContext.getRegistry());
if ("true".equals(element.getAttribute(ASYNC_EVENT_MULTICASTER_ATTRIBUTE).toLowerCase())) {
if (taskExecutor == null) {
taskExecutor = this.createTaskExecutor(1, 10, 0, "event-multicaster-");
}
TaskExecutor taskExecutor = IntegrationContextUtils.createTaskExecutor(1, 10, 0, "event-multicaster-");
BeanDefinitionBuilder eventMulticasterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
SimpleApplicationEventMulticaster.class);
eventMulticasterBuilder.addPropertyValue("taskExecutor", taskExecutor);
@@ -127,19 +85,6 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
this.addPostProcessors(element, parserContext);
}
private TaskExecutor createTaskExecutor(int coreSize, int maxSize, int queueCapacity, String threadPrefix) {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(coreSize);
executor.setMaxPoolSize(maxSize);
executor.setQueueCapacity(queueCapacity);
if (StringUtils.hasText(threadPrefix)) {
executor.setThreadFactory(new CustomizableThreadFactory(threadPrefix));
}
executor.setRejectedExecutionHandler(new CallerRunsPolicy());
executor.afterPropertiesSet();
return executor;
}
/**
* Adds extra post-processors to the context, to inject the objects configured by the MessageBus
*/

3
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd
View File

@@ -24,8 +24,6 @@
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="enable-annotations" type="xsd:boolean"/>
<xsd:attribute name="task-scheduler" type="xsd:string"/>
<xsd:attribute name="auto-startup" type="xsd:boolean"/>
<xsd:attribute name="configure-async-event-multicaster" type="xsd:boolean"/>
</xsd:complexType>
</xsd:element>
@@ -175,6 +173,7 @@
<xsd:attribute name="channel" type="xsd:string"/>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string"/>
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
</xsd:complexType>
<xsd:element name="service-activator" type="inputOutputEndpointType">

78
org.springframework.integration/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* Copyright 2002-2008 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.context;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Mark Fisher
*/
public abstract class IntegrationContextUtils {
public static final String TASK_SCHEDULER_BEAN_NAME = "taskScheduler";
public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel";
public static MessageChannel getErrorChannel(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, ERROR_CHANNEL_BEAN_NAME, MessageChannel.class);
}
public static TaskScheduler getTaskScheduler(BeanFactory beanFactory) {
return getBeanOfType(beanFactory, TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class);
}
public static TaskScheduler getRequiredTaskScheduler(BeanFactory beanFactory) {
TaskScheduler taskScheduler = getTaskScheduler(beanFactory);
Assert.state(taskScheduler != null, "No such bean '" + TASK_SCHEDULER_BEAN_NAME + "'");
return taskScheduler;
}
@SuppressWarnings("unchecked")
private static <T> T getBeanOfType(BeanFactory beanFactory, String beanName, Class<T> type) {
if (!beanFactory.containsBean(beanName)) {
return null;
}
Object bean = beanFactory.getBean(beanName);
Assert.state(type.isAssignableFrom(bean.getClass()), "incorrect type for bean '" + beanName
+ "' expected [" + type + "], but actual type is [" + bean.getClass() + "].");
return (T) bean;
}
public static TaskExecutor createTaskExecutor(int coreSize, int maxSize, int queueCapacity, String threadPrefix) {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(coreSize);
executor.setMaxPoolSize(maxSize);
executor.setQueueCapacity(queueCapacity);
if (StringUtils.hasText(threadPrefix)) {
executor.setThreadFactory(new CustomizableThreadFactory(threadPrefix));
}
executor.setRejectedExecutionHandler(new CallerRunsPolicy());
executor.afterPropertiesSet();
return executor;
}
}

47
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -16,29 +16,58 @@
package org.springframework.integration.endpoint;
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.BeanNameAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.util.LifecycleSupport;
import org.springframework.util.Assert;
/**
* The base class for Message Endpoint implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware {
public abstract class AbstractEndpoint extends LifecycleSupport implements MessageEndpoint, BeanNameAware, BeanFactoryAware {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile String beanName;
private volatile String name;
private volatile BeanFactory beanFactory;
private volatile TaskScheduler taskScheduler;
public void setBeanName(String name) {
this.name = name;
public void setBeanName(String beanName) {
this.beanName = beanName;
}
// TODO: make this final (see TODO in GatewayProxyFactoryBean)
public void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "beanFactory must not be null");
this.beanFactory = beanFactory;
TaskScheduler taskScheduler = IntegrationContextUtils.getTaskScheduler(beanFactory);
if (taskScheduler != null) {
this.setTaskScheduler(taskScheduler);
}
}
protected BeanFactory getBeanFactory() {
return this.beanFactory;
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
Assert.notNull(taskScheduler, "taskScheduler must not be null");
this.taskScheduler = taskScheduler;
}
protected TaskScheduler getTaskScheduler() {
return this.taskScheduler;
}
@Override
public String toString() {
return (this.name != null) ? this.name : super.toString();
return (this.beanName != null) ? this.beanName : super.toString();
}
}

58
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -25,11 +25,8 @@ import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.TaskSchedulerAware;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
@@ -43,8 +40,7 @@ import org.springframework.util.ClassUtils;
/**
* @author Mark Fisher
*/
public abstract class AbstractPollingEndpoint extends AbstractEndpoint
implements TaskSchedulerAware, Lifecycle, InitializingBean, BeanClassLoaderAware {
public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements InitializingBean, BeanClassLoaderAware {
public static final int MAX_MESSAGES_UNBOUNDED = -1;
@@ -65,15 +61,13 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint
private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
private volatile TaskScheduler taskScheduler;
private volatile ScheduledFuture<?> runningTask;
private volatile Runnable poller;
private volatile boolean initialized;
private final Object lifecycleMonitor = new Object();
private final Object initializationMonitor = new Object();
public void setTrigger(Trigger trigger) {
@@ -93,10 +87,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@@ -126,13 +116,14 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint
private TransactionTemplate getTransactionTemplate() {
if (!this.initialized) {
this.afterPropertiesSet();
this.onInit();
}
return this.transactionTemplate;
}
public void afterPropertiesSet() {
synchronized (this.lifecycleMonitor) {
@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
@@ -163,35 +154,24 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint
}
// Lifecycle implementation
// LifecycleSupport implementation
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.runningTask != null;
@Override // guarded by super#lifecycleLock
protected void doStart() {
if (!this.initialized) {
this.onInit();
}
Assert.state(this.getTaskScheduler() != null,
"unable to start polling, no taskScheduler available");
this.runningTask = this.getTaskScheduler().schedule(this.poller, this.trigger);
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (!this.initialized) {
this.afterPropertiesSet();
}
if (this.isRunning()) {
return;
}
Assert.state(this.taskScheduler != null,
"unable to start polling, no taskScheduler available");
this.runningTask = this.taskScheduler.schedule(this.poller, this.trigger);
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.runningTask != null) {
this.runningTask.cancel(true);
}
this.runningTask = null;
@Override // guarded by super#lifecycleLock
protected void doStop() {
if (this.runningTask != null) {
this.runningTask.cancel(true);
}
this.runningTask = null;
}

33
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
View File

@@ -16,7 +16,6 @@
package org.springframework.integration.endpoint;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
@@ -27,16 +26,12 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class EventDrivenConsumer extends AbstractEndpoint implements Lifecycle {
public class EventDrivenConsumer extends AbstractEndpoint {
private final SubscribableChannel inputChannel;
private final MessageHandler handler;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
@@ -46,28 +41,14 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Lifecycle {
}
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.running;
}
@Override // guarded by super#lifecycleLock
protected void doStart() {
this.inputChannel.subscribe(this.handler);
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
this.inputChannel.subscribe(this.handler);
this.running = true;
}
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.running) {
this.inputChannel.unsubscribe(this.handler);
this.running = false;
}
}
@Override // guarded by super#lifecycleLock
protected void doStop() {
this.inputChannel.unsubscribe(this.handler);
}
}

6
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -16,7 +16,6 @@
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
@@ -28,7 +27,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class MessageProducerSupport extends AbstractEndpoint implements InitializingBean {
public abstract class MessageProducerSupport extends AbstractEndpoint {
private volatile MessageChannel outputChannel;
@@ -43,7 +42,8 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
this.channelTemplate.setSendTimeout(sendTimeout);
}
public void afterPropertiesSet() {
@Override
protected void onInit() {
Assert.notNull(this.outputChannel, "outputChannel is required");
}

5
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -59,15 +59,16 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint {
this.channelTemplate.setSendTimeout(sendTimeout);
}
public void afterPropertiesSet() {
@Override
protected void onInit() {
Assert.notNull(this.source, "source must not be null");
Assert.notNull(this.outputChannel, "outputChannel must not be null");
super.afterPropertiesSet();
if (this.maxMessagesPerPoll < 0) {
// the default is 1 since a source might return
// a non-null value every time it is invoked
this.setMaxMessagesPerPoll(1);
}
super.onInit();
}
@Override

41
org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
View File

@@ -23,6 +23,7 @@ import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -31,8 +32,6 @@ import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.TaskSchedulerAware;
import org.springframework.util.Assert;
/**
@@ -43,7 +42,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class AbstractMessagingGateway implements MessagingGateway, MessageEndpoint, TaskSchedulerAware, Lifecycle {
public abstract class AbstractMessagingGateway extends AbstractEndpoint implements MessagingGateway {
private volatile MessageChannel requestChannel;
@@ -53,16 +52,10 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
private volatile boolean shouldThrowErrors = true;
private volatile TaskScheduler taskScheduler;
private volatile MessageEndpoint replyMessageCorrelator;
private final Object replyMessageCorrelatorMonitor = new Object();
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
/**
* Set the request channel.
@@ -113,10 +106,6 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
this.shouldThrowErrors = shouldThrowErrors;
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
public void send(Object object) {
Assert.state(this.requestChannel != null,
"send is not supported, because no request channel has been configured");
@@ -190,7 +179,7 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
else if (this.replyChannel instanceof PollableChannel) {
PollingConsumer endpoint = new PollingConsumer(
(PollableChannel) this.replyChannel, handler);
endpoint.setTaskScheduler(this.taskScheduler);
endpoint.setBeanFactory(this.getBeanFactory());
endpoint.afterPropertiesSet();
correlator = endpoint;
}
@@ -201,25 +190,17 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
}
}
public boolean isRunning() {
return this.running;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) {
((Lifecycle) this.replyMessageCorrelator).start();
}
this.running = true;
@Override // guarded by super#lifecycleLock
protected void doStart() {
if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) {
((Lifecycle) this.replyMessageCorrelator).start();
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) {
((Lifecycle) this.replyMessageCorrelator).stop();
}
this.running = false;
@Override // guarded by super#lifecycleLock
protected void doStop() {
if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) {
((Lifecycle) this.replyMessageCorrelator).stop();
}
}

58
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -37,11 +37,10 @@ import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.message.MethodParameterMessageMapper;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
@@ -52,8 +51,8 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
*/
public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, BeanClassLoaderAware, BeanFactoryAware,
InitializingBean, Lifecycle {
public class GatewayProxyFactoryBean extends AbstractEndpoint implements FactoryBean, MethodInterceptor, BeanClassLoaderAware, BeanFactoryAware,
InitializingBean {
private volatile Class<?> serviceInterface;
@@ -73,18 +72,12 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
private final Map<Method, MessagingGateway> gatewayMap = new HashMap<Method, MessagingGateway>();
private volatile TaskScheduler taskScheduler;
private volatile ChannelResolver channelResolver;
private volatile boolean initialized;
private volatile boolean running;
private final Object initializationMonitor = new Object();
private final Object lifecycleMonitor = new Object();
public void setServiceInterface(Class<?> serviceInterface) {
if (serviceInterface != null && !serviceInterface.isInterface()) {
@@ -144,12 +137,14 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
this.beanClassLoader = beanClassLoader;
}
@Override // TODO: remove this and move channelResolver to parent class
public void setBeanFactory(BeanFactory beanFactory) {
this.taskScheduler = (TaskScheduler) beanFactory.getBean(MessageBusParser.TASK_SCHEDULER_BEAN_NAME);
super.setBeanFactory(beanFactory);
this.channelResolver = new BeanFactoryChannelResolver(beanFactory);
}
public void afterPropertiesSet() throws Exception {
@Override
protected void onInit() throws Exception {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
@@ -226,7 +221,9 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
private MessagingGateway createGatewayForMethod(Method method) throws Exception {
SimpleMessagingGateway gateway = new SimpleMessagingGateway(
new MethodParameterMessageMapper(method), new SimpleMessageMapper());
gateway.setTaskScheduler(this.taskScheduler);
if (this.getTaskScheduler() != null) {
gateway.setTaskScheduler(this.getTaskScheduler());
}
Gateway gatewayAnnotation = method.getAnnotation(Gateway.class);
MessageChannel requestChannel = this.defaultRequestChannel;
MessageChannel replyChannel = this.defaultReplyChannel;
@@ -251,37 +248,28 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor,
gateway.setReplyChannel(replyChannel);
gateway.setRequestTimeout(requestTimeout);
gateway.setReplyTimeout(replyTimeout);
if (this.getBeanFactory() != null) {
gateway.setBeanFactory(this.getBeanFactory());
}
return gateway;
}
// Lifecycle implementation
public boolean isRunning() {
return this.running;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
for (MessagingGateway gateway : this.gatewayMap.values()) {
if (gateway instanceof Lifecycle) {
((Lifecycle) gateway).start();
}
}
this.running = true;
@Override // guarded by super#lifecycleLock
protected void doStart() {
for (MessagingGateway gateway : this.gatewayMap.values()) {
if (gateway instanceof Lifecycle) {
((Lifecycle) gateway).start();
}
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.running) {
for (MessagingGateway gateway : this.gatewayMap.values()) {
if (gateway instanceof Lifecycle) {
((Lifecycle) gateway).stop();
}
}
this.running = false;
@Override // guarded by super#lifecycleLock
protected void doStop() {
for (MessagingGateway gateway : this.gatewayMap.values()) {
if (gateway instanceof Lifecycle) {
((Lifecycle) gateway).stop();
}
}
}

84
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleTaskScheduler.java
View File

@@ -26,14 +26,18 @@ import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.integration.util.LifecycleSupport;
import org.springframework.scheduling.SchedulingException;
import org.springframework.util.Assert;
@@ -43,7 +47,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class SimpleTaskScheduler implements TaskScheduler, DisposableBean {
public class SimpleTaskScheduler extends LifecycleSupport implements TaskScheduler, BeanFactoryAware, DisposableBean {
private final Log logger = LogFactory.getLog(this.getClass());
@@ -57,14 +61,11 @@ public class SimpleTaskScheduler implements TaskScheduler, DisposableBean {
private final Set<TriggeredTask<?>> executingTasks = Collections.synchronizedSet(new TreeSet<TriggeredTask<?>>());
private volatile boolean running;
private final ReentrantLock lifecycleLock = new ReentrantLock();
public SimpleTaskScheduler(TaskExecutor executor) {
Assert.notNull(executor, "executor must not be null");
this.executor = executor;
this.setAutoStartMode(AutoStartMode.ON_CONTEXT_REFRESH);
}
@@ -72,7 +73,14 @@ public class SimpleTaskScheduler implements TaskScheduler, DisposableBean {
this.errorHandler = errorHandler;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
}
}
public final ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
Assert.notNull(task, "task must not be null");
TriggeredTask<Void> triggeredTask = new TriggeredTask<Void>(task, trigger);
return this.schedule(triggeredTask, null, null);
}
@@ -87,56 +95,28 @@ public class SimpleTaskScheduler implements TaskScheduler, DisposableBean {
}
// Lifecycle implementation
// LifecycleSupport implementation
public boolean isRunning() {
this.lifecycleLock.lock();
try {
return this.running;
}
finally {
this.lifecycleLock.unlock();
}
@Override // guarded by super#lifecycleLock
protected void doStart() {
this.executor.execute(this.schedulerTask = new SchedulerTask());
}
public void start() {
this.lifecycleLock.lock();
try {
if (this.running) {
return;
}
this.running = true;
this.executor.execute(this.schedulerTask = new SchedulerTask());
@Override // guarded by super#lifecycleLock
protected void doStop() {
this.schedulerTask.deactivate();
Thread executingThread = this.schedulerTask.executingThread.get();
if (executingThread != null) {
executingThread.interrupt();
}
finally {
this.lifecycleLock.unlock();
}
}
public void stop() {
this.lifecycleLock.lock();
try {
if (!this.running) {
return;
this.scheduledTasks.clear();
synchronized (this.executingTasks) {
for (TriggeredTask<?> task : this.executingTasks) {
task.cancel(true);
}
this.running = false;
this.schedulerTask.deactivate();
Thread executingThread = this.schedulerTask.executingThread.get();
if (executingThread != null) {
executingThread.interrupt();
}
this.scheduledTasks.clear();
synchronized (this.executingTasks) {
for (TriggeredTask<?> task : this.executingTasks) {
task.cancel(true);
}
this.executingTasks.clear();
}
this.schedulerTask = null;
}
finally {
this.lifecycleLock.unlock();
this.executingTasks.clear();
}
this.schedulerTask = null;
}
public void destroy() throws Exception {
@@ -236,7 +216,7 @@ public class SimpleTaskScheduler implements TaskScheduler, DisposableBean {
public long getDelay(TimeUnit unit) {
long now = new Date().getTime();
long scheduled = this.scheduledTime.getTime();
long scheduled = (this.scheduledTime != null) ? this.scheduledTime.getTime() : now;
return (scheduled > now) ? unit.convert(scheduled - now, TimeUnit.MILLISECONDS) : 0;
}

142
org.springframework.integration/src/main/java/org/springframework/integration/util/LifecycleSupport.java Normal file
View File

@@ -0,0 +1,142 @@
/*
* Copyright 2002-2008 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.util;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.event.ContextRefreshedEvent;
/**
* A convenience base class for Lifecycle components that supports an
* "auto-startup" mode property. Depending on the mode, the component can
* be started either upon initialization, upon receiving the
* {@link ContextRefreshedEvent}, or may require an explicit start invocation.
* The timing of the startup is determined by the value of {@link #autoStartMode}.
* The default value is {@link AutoStartMode#ON_INIT}. To require explicit startup,
* set the mode to {@link AutoStartMode#NONE} using the
* {@link #setAutoStartMode(AutoStartMode)} method.
*
* @author Mark Fisher
*/
public abstract class LifecycleSupport implements Lifecycle, InitializingBean, ApplicationListener {
public static enum AutoStartMode { ON_INIT, ON_CONTEXT_REFRESH, NONE }
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile AutoStartMode autoStartMode = AutoStartMode.ON_INIT;
private volatile boolean running;
private final ReentrantLock lifecycleLock = new ReentrantLock();
public void setAutoStartMode(AutoStartMode autoStartMode) {
this.autoStartMode = (autoStartMode != null) ? autoStartMode : AutoStartMode.NONE;
}
public final void afterPropertiesSet() {
try {
this.onInit();
if (this.autoStartMode == AutoStartMode.ON_INIT) {
this.start();
}
}
catch (Exception e) {
throw new BeanInitializationException("failed to initialize", e);
}
}
public final void onApplicationEvent(ApplicationEvent event) {
this.onEvent(event);
if (event instanceof ContextRefreshedEvent && this.autoStartMode == AutoStartMode.ON_CONTEXT_REFRESH) {
this.start();
}
}
// Lifecycle implementation
public final boolean isRunning() {
this.lifecycleLock.lock();
try {
return this.running;
}
finally {
this.lifecycleLock.unlock();
}
}
public final void start() {
this.lifecycleLock.lock();
try {
if (!this.running) {
this.doStart();
this.running = true;
if (logger.isInfoEnabled()) {
logger.info("started " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
}
}
public final void stop() {
this.lifecycleLock.lock();
try {
if (this.running) {
this.doStop();
this.running = false;
if (logger.isInfoEnabled()) {
logger.info("stopped " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
}
}
protected void onInit() throws Exception {
}
protected void onEvent(ApplicationEvent event) {
}
/**
* Subclasses must implement this method with the start behavior.
* This method will be invoked while holding the {@link #lifecycleLock}.
*/
protected abstract void doStart();
/**
* Subclasses must implement this method with the stop behavior.
* This method will be invoked while holding the {@link #lifecycleLock}.
*/
protected abstract void doStop();
}