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

INT-845 added StartupMode enum since polling consumers must not start as eagerly as event-driven consumers subscribe to their input channels

Browse Source
This commit is contained in:
Mark Fisher
2009-10-13 00:39:30 +00:00
parent 78da00302e
commit 2064939c84
23 changed files with 259 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<ContextRefreshedEvent> {
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);
}
}

19
org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -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<ContextRefreshedEvent> {
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);

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

@@ -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 + "]");
}
}
}
}

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

@@ -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() {

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

@@ -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;
}

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

@@ -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;
}

3
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -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;
}

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

@@ -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();
}
}

3
org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java
View File

@@ -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());

2
org.springframework.integration/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java
View File

@@ -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());

5
org.springframework.integration/src/test/java/org/springframework/integration/config/xml/MethodInvokingOutboundChannelAdapterParserTests.java
View File

@@ -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"));
}

24
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml Normal file
View File

@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
<si:inbound-channel-adapter id="inboundEndpoint" ref="counter" method="next" channel="testChannel">
<si:poller max-messages-per-poll="1">
<si:interval-trigger interval="50"/>
</si:poller>
</si:inbound-channel-adapter>
<si:channel id="testChannel"/>
<si:service-activator id="consumerEndpoint" input-channel="testChannel" ref="consumer"/>
<bean id="counter" class="org.springframework.integration.endpoint.ProducerAndConsumerAutoStartupTests$Counter"/>
<bean id="consumer" class="org.springframework.integration.endpoint.ProducerAndConsumerAutoStartupTests$Consumer"/>
</beans>

87
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests.java Normal file
View File

@@ -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<Integer> received = new ArrayList<Integer>();
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<Integer> numbers = new LinkedBlockingQueue<Integer>();
public void receive(Integer number) {
numbers.add(number);
}
Integer poll(long timeoutInMillis) throws InterruptedException {
return numbers.poll(timeoutInMillis, TimeUnit.MILLISECONDS);
}
}
}