From 2064939c84d9337b2661f0904cc1566a6c77e975 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 13 Oct 2009 00:39:30 +0000 Subject: [PATCH] INT-845 added StartupMode enum since polling consumers must not start as eagerly as event-driven consumers subscribe to their input channels --- ...ApplicationEventInboundChannelAdapter.java | 3 +- ...dChannelAdapterWithPatternParserTests.java | 4 +- ...FileOutboundChannelAdapterParserTests.java | 3 +- .../FileOutboundGatewayParserTests.java | 3 +- .../HttpOutboundGatewayParserTests.java | 3 +- .../jms/JmsMessageDrivenEndpoint.java | 14 ++- .../mail/ImapIdleChannelAdapter.java | 16 ++++ .../config/PollingMailSourceParserTests.java | 5 +- .../SimpleWebServiceInboundGatewayTests.java | 3 + .../WebServiceOutboundGatewayParserTests.java | 3 +- .../config/ConsumerEndpointFactoryBean.java | 17 ++-- ...ourcePollingChannelAdapterFactoryBean.java | 19 +++- .../MessagingAnnotationPostProcessor.java | 17 +++- .../endpoint/AbstractEndpoint.java | 28 ++++-- .../endpoint/AbstractPollingEndpoint.java | 5 ++ .../endpoint/EventDrivenConsumer.java | 3 +- .../integration/endpoint/PollingConsumer.java | 3 +- .../gateway/AbstractMessagingGateway.java | 30 +++++-- .../channel/MessageChannelTemplateTests.java | 3 + ...PollingChannelAdapterFactoryBeanTests.java | 2 + ...kingOutboundChannelAdapterParserTests.java | 5 +- ...cerAndConsumerAutoStartupTests-context.xml | 24 +++++ .../ProducerAndConsumerAutoStartupTests.java | 87 +++++++++++++++++++ 23 files changed, 259 insertions(+), 41 deletions(-) create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests.java diff --git a/org.springframework.integration.event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java b/org.springframework.integration.event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java index 524f67f44c..174399626d 100644 --- a/org.springframework.integration.event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java +++ b/org.springframework.integration.event/src/main/java/org/springframework/integration/event/ApplicationEventInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -51,6 +51,7 @@ public class ApplicationEventInboundChannelAdapter extends MessageProducerSuppor } public void onApplicationEvent(ApplicationEvent event) { + super.onApplicationEvent(event); if (CollectionUtils.isEmpty(this.eventTypes)) { this.sendEventAsMessage(event); return; diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithPatternParserTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithPatternParserTests.java index d09e2aa79b..aed0693e16 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithPatternParserTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithPatternParserTests.java @@ -34,6 +34,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.file.AcceptOnceFileListFilter; import org.springframework.integration.file.CompositeFileListFilter; import org.springframework.integration.file.FileListFilter; @@ -73,8 +74,7 @@ public class FileInboundChannelAdapterWithPatternParserTests { @Test public void autoStartupDisabled() { assertFalse(this.endpoint.isRunning()); - Boolean autoStartupValue = (Boolean) new DirectFieldAccessor(endpoint).getPropertyValue("autoStartup"); - assertFalse(autoStartupValue); + assertEquals(StartupMode.MANUAL, new DirectFieldAccessor(endpoint).getPropertyValue("startupMode")); } @Test diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java index d9ad985193..fe65450642 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java @@ -28,6 +28,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.FileWritingMessageHandler; import org.springframework.test.context.ContextConfiguration; @@ -104,7 +105,7 @@ public class FileOutboundChannelAdapterParserTests { @Test public void adapterWithAutoStartupFalse() { DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapterWithOrder); - assertEquals(Boolean.FALSE, adapterAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, adapterAccessor.getPropertyValue("startupMode")); } } diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java index 574176a00c..c07f6c0850 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java @@ -24,6 +24,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.file.FileWritingMessageHandler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -45,7 +46,7 @@ public class FileOutboundGatewayParserTests { DirectFieldAccessor gatewayAccessor = new DirectFieldAccessor(gateway); FileWritingMessageHandler handler = (FileWritingMessageHandler) gatewayAccessor.getPropertyValue("handler"); - assertEquals(Boolean.FALSE, gatewayAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, gatewayAccessor.getPropertyValue("startupMode")); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); assertEquals(777, handlerAccessor.getPropertyValue("order")); } diff --git a/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java b/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java index a5e2b97040..bf8fcad7f0 100644 --- a/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java +++ b/org.springframework.integration.http/src/test/java/org/springframework/integration/http/config/HttpOutboundGatewayParserTests.java @@ -33,6 +33,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.http.DefaultOutboundRequestMapper; import org.springframework.integration.http.HttpOutboundEndpoint; import org.springframework.integration.http.HttpRequestExecutor; @@ -92,7 +93,7 @@ public class HttpOutboundGatewayParserTests { assertEquals(this.applicationContext.getBean("requests"), requestChannel); DirectFieldAccessor accessor = new DirectFieldAccessor(gateway); assertEquals(77, accessor.getPropertyValue("order")); - assertEquals(Boolean.FALSE, endpointAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, endpointAccessor.getPropertyValue("startupMode")); Object replyChannel = accessor.getPropertyValue("outputChannel"); assertNotNull(replyChannel); assertEquals(this.applicationContext.getBean("replies"), replyChannel); diff --git a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java index e384f06c68..da2c101ad3 100644 --- a/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java +++ b/org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -33,6 +33,8 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos private final ChannelPublishingJmsMessageListener listener; + private volatile boolean autoStartup = true; + public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, ChannelPublishingJmsMessageListener listener) { Assert.notNull(listenerContainer, "listener container must not be null"); @@ -43,8 +45,18 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements Dispos } + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + @Override protected void onInit() throws Exception { + if (this.autoStartup) { + this.setStartupMode(StartupMode.ON_CONTEXT_REFRESH); + } + else { + this.setStartupMode(StartupMode.MANUAL); + } this.listener.afterPropertiesSet(); if (!this.listenerContainer.isActive()) { this.listenerContainer.afterPropertiesSet(); diff --git a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java index afd0889ca5..e1f1f15985 100755 --- a/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java +++ b/org.springframework.integration.mail/src/main/java/org/springframework/integration/mail/ImapIdleChannelAdapter.java @@ -41,6 +41,8 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { private final IdleTask idleTask = new IdleTask(); + private volatile boolean autoStartup = true; + private volatile boolean shouldReconnectAutomatically = true; private volatile TaskExecutor taskExecutor; @@ -63,6 +65,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { this.shouldReconnectAutomatically = shouldReconnectAutomatically; } + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + public void setTaskExecutor(TaskExecutor taskExecutor) { this.taskExecutor = taskExecutor; } @@ -73,6 +79,16 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport { } } + @Override + protected void onInit() { + if (this.autoStartup) { + this.setStartupMode(StartupMode.ON_CONTEXT_REFRESH); + } + else { + this.setStartupMode(StartupMode.MANUAL); + } + } + /* * Lifecycle implementation */ diff --git a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/PollingMailSourceParserTests.java b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/PollingMailSourceParserTests.java index 7707976ca9..6384baa1a9 100644 --- a/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/PollingMailSourceParserTests.java +++ b/org.springframework.integration.mail/src/test/java/org/springframework/integration/mail/config/PollingMailSourceParserTests.java @@ -29,6 +29,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.mail.ImapMailReceiver; import org.springframework.integration.mail.MailReceivingMessageSource; import org.springframework.integration.mail.Pop3MailReceiver; @@ -52,7 +53,7 @@ public class PollingMailSourceParserTests { Object adapter = context.getBean("imapAdapter"); assertEquals(SourcePollingChannelAdapter.class, adapter.getClass()); DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); - assertEquals(Boolean.FALSE, adapterAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, adapterAccessor.getPropertyValue("startupMode")); Object channel = context.getBean("channel"); assertEquals(channel, adapterAccessor.getPropertyValue("outputChannel")); Object source = adapterAccessor.getPropertyValue("source"); @@ -71,7 +72,7 @@ public class PollingMailSourceParserTests { Object adapter = context.getBean("pop3Adapter"); assertEquals(SourcePollingChannelAdapter.class, adapter.getClass()); DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); - assertEquals(Boolean.FALSE, adapterAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, adapterAccessor.getPropertyValue("startupMode")); Object channel = context.getBean("channel"); assertEquals(channel, adapterAccessor.getPropertyValue("outputChannel")); Object source = adapterAccessor.getPropertyValue("source"); diff --git a/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceInboundGatewayTests.java b/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceInboundGatewayTests.java index 4b6d9d61f3..0915fc87a9 100644 --- a/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceInboundGatewayTests.java +++ b/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/SimpleWebServiceInboundGatewayTests.java @@ -37,6 +37,8 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.runners.MockitoJUnit44Runner; import org.mockito.stubbing.Answer; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; @@ -90,6 +92,7 @@ public class SimpleWebServiceInboundGatewayTests { when(requestChannel.send(isA(Message.class))).thenAnswer( withReplyTo(replyChannel)); when(request.getPayloadSource()).thenReturn(payloadSource); + gateway.onApplicationEvent(new ContextRefreshedEvent(new StaticApplicationContext())); gateway.invoke(context); verify(requestChannel).send(messageWithPayload(payloadSource)); assertTrue(output.toString().endsWith(input)); diff --git a/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java b/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java index 9b94027626..c9f1ded565 100644 --- a/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java +++ b/org.springframework.integration.ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java @@ -27,6 +27,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.ws.MarshallingWebServiceOutboundGateway; import org.springframework.integration.ws.SimpleWebServiceOutboundGateway; import org.springframework.oxm.Marshaller; @@ -221,7 +222,7 @@ public class WebServiceOutboundGatewayParserTests { ApplicationContext context = new ClassPathXmlApplicationContext( "simpleWebServiceOutboundGatewayParserTests.xml", this.getClass()); AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithOrderAndAutoStartupFalse"); - assertEquals(Boolean.FALSE, new DirectFieldAccessor(endpoint).getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, new DirectFieldAccessor(endpoint).getPropertyValue("startupMode")); } @Test diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 4fec45f139..ae3a6bba2a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -22,9 +22,9 @@ 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.context.Lifecycle; +import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; import org.springframework.integration.context.IntegrationContextUtils; @@ -32,6 +32,7 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.util.Assert; @@ -40,7 +41,7 @@ import org.springframework.util.Assert; * @author Mark Fisher */ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, - InitializingBean, Lifecycle, ApplicationListener { + InitializingBean, Lifecycle, ApplicationListener { private volatile MessageHandler handler; @@ -152,12 +153,12 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar throw new IllegalArgumentException( "unsupported channel type: [" + channel.getClass() + "]"); } - this.endpoint.setAutoStartup(this.autoStartup); this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); - if (this.endpoint instanceof InitializingBean) { - ((InitializingBean) this.endpoint).afterPropertiesSet(); + if (!this.autoStartup) { + this.endpoint.setStartupMode(StartupMode.MANUAL); } + this.endpoint.afterPropertiesSet(); this.initialized = true; } } @@ -186,10 +187,8 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar * ApplicationListener implementation */ - public void onApplicationEvent(ApplicationEvent event) { - if (this.endpoint instanceof ApplicationListener) { - ((ApplicationListener) this.endpoint).onApplicationEvent(event); - } + public void onApplicationEvent(ContextRefreshedEvent event) { + this.endpoint.onApplicationEvent(event); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index 94d65c3b25..b7885c7d19 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -23,10 +23,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.ApplicationListener; import org.springframework.context.Lifecycle; +import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.message.MessageSource; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.util.Assert; @@ -36,8 +39,8 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class SourcePollingChannelAdapterFactoryBean - implements FactoryBean, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, Lifecycle { +public class SourcePollingChannelAdapterFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, + BeanClassLoaderAware, InitializingBean, Lifecycle, ApplicationListener { private volatile MessageSource source; @@ -94,6 +97,12 @@ public class SourcePollingChannelAdapterFactoryBean this.initializeAdapter(); } + public void onApplicationEvent(ContextRefreshedEvent event) { + if (this.adapter != null) { + this.adapter.onApplicationEvent(event); + } + } + public Object getObject() throws Exception { if (this.adapter == null) { this.initializeAdapter(); @@ -130,7 +139,9 @@ public class SourcePollingChannelAdapterFactoryBean spca.setTransactionManager(this.pollerMetadata.getTransactionManager()); spca.setTransactionDefinition(this.pollerMetadata.getTransactionDefinition()); spca.setAdviceChain(this.pollerMetadata.getAdviceChain()); - spca.setAutoStartup(this.autoStartup); + if (!this.autoStartup) { + spca.setStartupMode(StartupMode.MANUAL); + } spca.setBeanName(this.beanName); spca.setBeanFactory(this.beanFactory); spca.setBeanClassLoader(this.beanClassLoader); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index a9fb119b99..7a2f94071c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -26,6 +26,9 @@ import java.util.List; import java.util.Map; import java.util.Set; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -59,6 +62,8 @@ import org.springframework.util.StringUtils; */ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean, ApplicationListener { + private final Log logger = LogFactory.getLog(this.getClass()); + private volatile ConfigurableListableBeanFactory beanFactory; @@ -168,7 +173,15 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean public void onApplicationEvent(ApplicationEvent event) { for (ApplicationListener listener : listeners) { - listener.onApplicationEvent(event); + try { + listener.onApplicationEvent(event); + } + catch (ClassCastException e) { + if (logger.isWarnEnabled() && event != null) { + logger.warn("ApplicationEvent of type [" + event.getClass() + + "] not accepted by ApplicationListener [" + listener + "]"); + } + } } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 8dd5d989ae..637432cffe 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -20,7 +20,10 @@ import java.util.concurrent.locks.ReentrantLock; 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; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.scheduling.TaskScheduler; @@ -36,17 +39,25 @@ import org.springframework.scheduling.TaskScheduler; * * @author Mark Fisher */ -public abstract class AbstractEndpoint extends IntegrationObjectSupport implements Lifecycle, InitializingBean { +public abstract class AbstractEndpoint extends IntegrationObjectSupport + implements ApplicationListener, Lifecycle, InitializingBean { - private volatile boolean autoStartup = true; + public static enum StartupMode { + MANUAL, + ON_INITIALIZATION, + ON_CONTEXT_REFRESH; + } + + + private volatile StartupMode startupMode = StartupMode.MANUAL; private volatile boolean running; private final ReentrantLock lifecycleLock = new ReentrantLock(); - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; + public void setStartupMode(StartupMode startupMode) { + this.startupMode = (startupMode != null ? startupMode : StartupMode.MANUAL); } public void setTaskScheduler(TaskScheduler taskScheduler) { @@ -56,7 +67,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen public final void afterPropertiesSet() { try { this.onInit(); - if (this.autoStartup) { + if (this.startupMode == StartupMode.ON_INITIALIZATION) { this.start(); } } @@ -65,6 +76,13 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen } } + public void onApplicationEvent(ApplicationEvent event) { + if (event instanceof ContextRefreshedEvent + && this.startupMode == StartupMode.ON_CONTEXT_REFRESH) { + this.start(); + } + } + // Lifecycle implementation public final boolean isRunning() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 7a80388af8..4879cf520d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -74,6 +74,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private final Object initializationMonitor = new Object(); + public AbstractPollingEndpoint() { + this.setStartupMode(StartupMode.ON_CONTEXT_REFRESH); + } + + public void setTrigger(Trigger trigger) { this.trigger = trigger; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java index 4f4d19dbae..36c628c22f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -36,6 +36,7 @@ public class EventDrivenConsumer extends AbstractEndpoint { public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler handler) { Assert.notNull(inputChannel, "inputChannel must not be null"); Assert.notNull(handler, "handler must not be null"); + this.setStartupMode(StartupMode.ON_INITIALIZATION); this.inputChannel = inputChannel; this.handler = handler; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 8011daec5d..919c07f134 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -39,6 +39,7 @@ public class PollingConsumer extends AbstractPollingEndpoint { public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) { Assert.notNull(inputChannel, "inputChannel must not be null"); Assert.notNull(handler, "handler must not be null"); + this.setStartupMode(StartupMode.ON_CONTEXT_REFRESH); this.inputChannel = inputChannel; this.handler = handler; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index b64ac55507..d6ed4b1628 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -16,7 +16,6 @@ package org.springframework.integration.gateway; -import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; @@ -52,6 +51,8 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen private volatile boolean shouldThrowErrors = true; + private volatile boolean autoStartup = true; + private volatile boolean initialized; private volatile AbstractEndpoint replyMessageCorrelator; @@ -108,8 +109,23 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen this.shouldThrowErrors = shouldThrowErrors; } + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + @Override protected void onInit() throws Exception { + if (this.autoStartup) { + if (this.requestChannel instanceof PollableChannel) { + this.setStartupMode(StartupMode.ON_CONTEXT_REFRESH); + } + else { + this.setStartupMode(StartupMode.ON_INITIALIZATION); + } + } + else { + this.setStartupMode(StartupMode.MANUAL); + } this.initialized = true; } @@ -208,8 +224,8 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen endpoint.afterPropertiesSet(); correlator = endpoint; } - if (this.isRunning() && correlator instanceof Lifecycle) { - ((Lifecycle) correlator).start(); + if (this.isRunning()) { + correlator.start(); } this.replyMessageCorrelator = correlator; } @@ -217,15 +233,15 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint implemen @Override // guarded by super#lifecycleLock protected void doStart() { - if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) { - ((Lifecycle) this.replyMessageCorrelator).start(); + if (this.replyMessageCorrelator != null) { + replyMessageCorrelator.start(); } } @Override // guarded by super#lifecycleLock protected void doStop() { - if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) { - ((Lifecycle) this.replyMessageCorrelator).stop(); + if (this.replyMessageCorrelator != null) { + this.replyMessageCorrelator.stop(); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java index 42742f6227..6f3d75d677 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java @@ -182,6 +182,7 @@ public class MessageChannelTemplateTests { @Test public void sendAndReceive() { MessageChannelTemplate template = new MessageChannelTemplate(); + template.setReceiveTimeout(3000); Message reply = template.sendAndReceive(new StringMessage("test"), this.requestChannel); assertEquals("TEST", reply.getPayload()); } @@ -189,6 +190,7 @@ public class MessageChannelTemplateTests { @Test public void sendAndReceiveWithDefaultChannel() { MessageChannelTemplate template = new MessageChannelTemplate(); + template.setReceiveTimeout(3000); template.setDefaultChannel(this.requestChannel); Message reply = template.sendAndReceive(new StringMessage("test")); assertEquals("TEST", reply.getPayload()); @@ -198,6 +200,7 @@ public class MessageChannelTemplateTests { public void sendAndReceiveWithExplicitChannelTakesPrecedenceOverDefault() { QueueChannel defaultChannel = new QueueChannel(); MessageChannelTemplate template = new MessageChannelTemplate(defaultChannel); + template.setReceiveTimeout(3000); Message message = new StringMessage("test"); Message reply = template.sendAndReceive(message, this.requestChannel); assertEquals("TEST", reply.getPayload()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java index eda5dcc0c3..522db86d18 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java @@ -31,6 +31,7 @@ import org.junit.Test; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; +import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollerMetadata; @@ -68,6 +69,7 @@ public class SourcePollingChannelAdapterFactoryBeanTests { factoryBean.setPollerMetadata(pollerMetadata); factoryBean.setAutoStartup(true); factoryBean.afterPropertiesSet(); + context.registerEndpoint("testPollingEndpoint", (AbstractEndpoint) factoryBean.getObject()); context.refresh(); Message message = outputChannel.receive(30000); assertEquals("test", message.getPayload()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/MethodInvokingOutboundChannelAdapterParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/MethodInvokingOutboundChannelAdapterParserTests.java index b90c37d21e..c89f63b5c7 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/MethodInvokingOutboundChannelAdapterParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/MethodInvokingOutboundChannelAdapterParserTests.java @@ -25,6 +25,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; +import org.springframework.integration.endpoint.AbstractEndpoint.StartupMode; import org.springframework.integration.handler.MethodInvokingMessageHandler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -48,7 +49,7 @@ public class MethodInvokingOutboundChannelAdapterParserTests { assertEquals(MethodInvokingMessageHandler.class, handler.getClass()); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); assertEquals(99, handlerAccessor.getPropertyValue("order")); - assertEquals(Boolean.FALSE, adapterAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, adapterAccessor.getPropertyValue("startupMode")); } @Test @@ -59,7 +60,7 @@ public class MethodInvokingOutboundChannelAdapterParserTests { assertEquals(MethodInvokingMessageHandler.class, handler.getClass()); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); assertEquals(99, handlerAccessor.getPropertyValue("order")); - assertEquals(Boolean.FALSE, adapterAccessor.getPropertyValue("autoStartup")); + assertEquals(StartupMode.MANUAL, adapterAccessor.getPropertyValue("startupMode")); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml new file mode 100644 index 0000000000..3dc062809a --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests.java new file mode 100644 index 0000000000..90a96dc6a8 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests.java @@ -0,0 +1,87 @@ +/* + * Copyright 2002-2009 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.endpoint; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + * @since 2.0.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ProducerAndConsumerAutoStartupTests { + + @Autowired + private AbstractApplicationContext context; + + @Autowired + private Consumer consumer; + + + @Test + public void test() throws Exception { + List received = new ArrayList(); + for (int i = 0; i < 3; i++) { + received.add(consumer.poll(500)); + } + context.stop(); + assertEquals(new Integer(1), received.get(0)); + assertEquals(new Integer(2), received.get(1)); + assertEquals(new Integer(3), received.get(2)); + } + + + static class Counter { + + private final AtomicInteger count = new AtomicInteger(); + + public Integer next() { + return new Integer(count.incrementAndGet()); + } + } + + + static class Consumer { + + private final BlockingQueue numbers = new LinkedBlockingQueue(); + + public void receive(Integer number) { + numbers.add(number); + } + + Integer poll(long timeoutInMillis) throws InterruptedException { + return numbers.poll(timeoutInMillis, TimeUnit.MILLISECONDS); + } + } + +}