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

MessageEndpoint no longer has any awareness of the Schedule - only its MessageSource. The source may be a PollingDispatcher, but the poller is always configured without the MessageEndpoint needing to provide the Schedule.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-30 23:01:38 +00:00
parent bdfc98f966
commit f96354671c
7 changed files with 29 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

@@ -306,9 +306,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
return;
}
if (source != null && source instanceof PollableSource) {
Schedule schedule = endpoint.getSchedule();
schedule = schedule != null ? schedule : this.defaultPollerSchedule;
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) source, schedule);
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) source, this.defaultPollerSchedule);
poller.subscribe(endpoint);
this.pollingDispatchers.add(poller);
this.taskScheduler.schedule(poller);

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

@@ -27,10 +27,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DelegatingIntroductionInterceptor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
@@ -97,18 +94,6 @@ public abstract class AbstractAnnotationMethodPostProcessor<T> implements Annota
return null;
}
protected Schedule extractSchedule(Class<?> originalBeanClass) {
PollingSchedule schedule = null;
Poller pollerAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Poller.class);
if (pollerAnnotation != null) {
schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
}
return schedule;
}
protected abstract T processMethod(Object bean, Method method, Annotation annotation);

5
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java
View File

@@ -44,7 +44,6 @@ import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.router.RouterMessageHandlerCreator;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.splitter.SplitterMessageHandlerCreator;
import org.springframework.integration.transformer.config.TransformerMessageHandlerCreator;
import org.springframework.util.StringUtils;
@@ -137,10 +136,6 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
if (StringUtils.hasText(inputChannelName)) {
endpoint.setInputChannelName(inputChannelName);
}
Schedule schedule = this.extractSchedule(originalBeanClass);
if (schedule != null) {
endpoint.setSchedule(schedule);
}
return endpoint;
}

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

@@ -43,6 +43,7 @@ import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* A {@link BeanPostProcessor} implementation that processes method-level
@@ -99,28 +100,28 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation);
if (endpoint != null) {
endpoint.setBeanName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint");
String inputChannelName = endpointAnnotation.input();
if (!StringUtils.hasText(inputChannelName)) {
continue;
}
MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName);
if (inputChannel == null) {
throw new ConfigurationException("unable to resolve input channel '" + inputChannelName + "'");
}
Poller pollerAnnotation = AnnotationUtils.findAnnotation(beanClass, Poller.class);
if (pollerAnnotation != null) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
String inputChannelName = endpointAnnotation.input();
MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName);
if (inputChannel != null) {
if (inputChannel instanceof PollableChannel) {
PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule);
poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
endpoint.setSource(poller);
}
else {
endpoint.setSource(inputChannel);
}
if (inputChannel instanceof PollableChannel) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule);
poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
endpoint.setSource(poller);
}
else {
endpoint.setInputChannelName(inputChannelName);
throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource");
}
endpoint.setSchedule(schedule);
}
this.messageBus.registerEndpoint(endpoint);
}

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

@@ -28,7 +28,6 @@ import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.util.ErrorHandler;
/**
@@ -46,8 +45,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
private MessageTarget target;
private volatile Schedule schedule;
private volatile ErrorHandler errorHandler;
private volatile ChannelRegistry channelRegistry;
@@ -82,14 +79,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
this.target = target;
}
public Schedule getSchedule() {
return this.schedule;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
protected ChannelRegistry getChannelRegistry() {
return this.channelRegistry;
}

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

@@ -19,7 +19,6 @@ package org.springframework.integration.endpoint;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.Schedule;
/**
* Base interface for message endpoints.
@@ -30,10 +29,6 @@ public interface MessageEndpoint extends MessageTarget, BeanNameAware {
String getName();
void setSchedule(Schedule schedule);
Schedule getSchedule();
void setSource(MessageSource<?> source);
MessageSource<?> getSource();

11
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.annotation.Order;
@@ -55,7 +56,9 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.DefaultEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
@@ -167,6 +170,9 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testChannelRegistryAwareBean() {
MessageBus messageBus = new DefaultMessageBus();
QueueChannel inputChannel = new QueueChannel();
inputChannel.setBeanName("inputChannel");
messageBus.registerChannel(inputChannel);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
ChannelRegistryAwareTestBean testBean = new ChannelRegistryAwareTestBean();
@@ -351,7 +357,10 @@ public class MessagingAnnotationPostProcessorTests {
AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation();
postProcessor.postProcessAfterInitialization(endpoint, "testBean");
DefaultEndpoint<?> processedEndpoint = (DefaultEndpoint<?>) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint");
Schedule schedule = processedEndpoint.getSchedule();
DirectFieldAccessor accessor = new DirectFieldAccessor(processedEndpoint);
MessageSource<?> source = (MessageSource<?>) accessor.getPropertyValue("source");
assertTrue(source instanceof SubscribableSource);
Schedule schedule = (Schedule) new DirectFieldAccessor(source).getPropertyValue("schedule");
assertEquals(PollingSchedule.class, schedule.getClass());
PollingSchedule pollingSchedule = (PollingSchedule) schedule;
assertEquals(1234, pollingSchedule.getPeriod());