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

Providing better separation between PollableSource and PollingSourceAdapter (work in progress).

Browse Source
This commit is contained in:
Mark Fisher
2008-04-16 16:31:24 +00:00
parent 19aad42c9e
commit b5e01b447a
37 changed files with 481 additions and 593 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
View File

@@ -18,132 +18,74 @@ package org.springframework.integration.adapter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import org.springframework.context.Lifecycle;
import org.springframework.integration.ConfigurationException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
import org.springframework.util.Assert;
/**
* A channel adapter that retrieves objects from a {@link PollableSource},
* delegates to a {@link MessageMapper} to create messages from those objects,
* A channel adapter that retrieves messages from a {@link PollableSource}
* and then sends the resulting messages to the provided {@link MessageChannel}.
*
* @author Mark Fisher
*/
public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements MessagingTaskSchedulerAware, Lifecycle {
public class PollingSourceAdapter extends AbstractSourceAdapter implements MessagingTask, InitializingBean {
private volatile PollableSource<T> source;
private final Log logger = LogFactory.getLog(this.getClass());
private volatile PollingSchedule schedule = new PollingSchedule(1000);
private final PollableSource<?> source;
private volatile MessagingTaskScheduler scheduler;
private final PollingSchedule schedule;
private volatile int maxMessagesPerTask = 1;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
private volatile boolean initialized;
/**
* Create a new adapter for the given source.
*/
public PollingSourceAdapter(PollableSource<T> source) {
this.setSource(source);
}
/**
* No-arg constructor for providing source after construction.
*/
public PollingSourceAdapter() {
}
public void setSource(PollableSource<T> source) {
Assert.notNull(source, "'source' must not be null");
public PollingSourceAdapter(PollableSource<?> source, MessageChannel channel, PollingSchedule schedule) {
super(channel);
Assert.notNull(source, "source must not be null");
Assert.notNull(schedule, "schedule must not be null");
this.source = source;
this.schedule = schedule;
}
public void setInitialDelay(long intialDelay) {
Assert.isTrue(intialDelay >= 0, "'intialDelay' must not be negative");
this.schedule.setInitialDelay(intialDelay);
}
public void setPeriod(long period) {
this.schedule.setPeriod(period);
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least one");
this.maxMessagesPerTask = maxMessagesPerTask;
}
public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) {
Assert.notNull(scheduler, "scheduler must not be null");
this.scheduler = scheduler;
public Schedule getSchedule() {
return this.schedule;
}
protected PollableSource<T> getSource() {
return this.source;
}
public boolean isRunning() {
return this.running;
}
@Override
protected void initialize() {
if (this.source == null) {
throw new ConfigurationException("source must not be null");
}
public void afterPropertiesSet() {
if (this.getChannel() instanceof SynchronousChannel) {
((SynchronousChannel) this.getChannel()).setSource(this.source);
}
this.initialized = true;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
return;
}
if (!this.isInitialized()) {
this.afterPropertiesSet();
}
if (this.scheduler == null) {
if (logger.isInfoEnabled()) {
logger.info("no task scheduler has been provided, will create one");
}
this.scheduler = new SimpleMessagingTaskScheduler(Executors.newSingleThreadScheduledExecutor());
}
this.running = true;
}
if (!this.scheduler.isRunning()) {
this.scheduler.start();
}
this.scheduler.schedule(new PollingSourceAdapterTask());
}
public void stop() {
this.running = false;
}
public List<Message<T>> poll(int limit) {
List<Message<T>> results = new ArrayList<Message<T>>();
public List<Message<?>> poll(int limit) {
List<Message<?>> results = new ArrayList<Message<?>>();
int count = 0;
while (count < limit) {
Message<T> message = this.source.poll();
Message<?> message = this.source.poll();
if (message == null) {
break;
}
@@ -153,44 +95,35 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
return results;
}
public int processMessages() {
if (!this.isRunning()) {
if (logger.isDebugEnabled()) {
logger.debug("source adapter not polling since it has not yet been started");
}
return 0;
protected boolean sendMessage(Message<?> message) {
if (!this.initialized) {
this.afterPropertiesSet();
}
int messagesProcessed = 0;
List<Message<T>> messages = this.poll(this.maxMessagesPerTask);
for (Message<T> message : messages) {
if (this.sendToChannel(message)) {
messagesProcessed++;
this.onSend(message);
boolean sent = super.sendToChannel(message);
if (this.source instanceof MessageDeliveryAware) {
if (sent) {
((MessageDeliveryAware) this.source).onSend(message);
}
else {
return messagesProcessed;
((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
}
}
return messagesProcessed;
return sent;
}
/**
* Callback method invoked after a message is sent to the channel.
* <p>
* Subclasses may override. The default implementation does nothing.
*/
protected void onSend(Message<T> sentMessage) {
}
private class PollingSourceAdapterTask implements MessagingTask {
public void run() {
processMessages();
public void run() {
int messagesProcessed = 0;
List<Message<?>> messages = this.poll(this.maxMessagesPerTask);
for (Message<?> message : messages) {
if (this.sendMessage(message)) {
messagesProcessed++;
}
else {
break;
}
}
public Schedule getSchedule() {
return schedule;
if (logger.isDebugEnabled()) {
logger.debug("polling source task processed " + messagesProcessed + " messages");
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -16,8 +16,6 @@
package org.springframework.integration.adapter;
import org.springframework.integration.channel.MessageChannel;
/**
* Base interface for source adapters.
*
@@ -25,6 +23,4 @@ import org.springframework.integration.channel.MessageChannel;
*/
public interface SourceAdapter {
void setChannel(MessageChannel channel);
}

6
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -50,8 +50,8 @@ import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
import org.springframework.integration.scheduling.Subscription;
@@ -355,8 +355,8 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
if (!this.initialized) {
this.initialize();
}
if (adapter instanceof MessagingTaskSchedulerAware) {
((MessagingTaskSchedulerAware) adapter).setMessagingTaskScheduler(this.taskScheduler);
if (adapter instanceof MessagingTask) {
this.taskScheduler.schedule((MessagingTask) adapter);
}
if (adapter instanceof Lifecycle) {
this.lifecycleSourceAdapters.add((Lifecycle) adapter);

26
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -30,6 +30,7 @@ import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
@@ -77,21 +78,22 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
if (this.isInbound) {
adapterDef = new RootBeanDefinition(PollingSourceAdapter.class);
invokerDef = new RootBeanDefinition(MethodInvokingSource.class);
String invokerBeanName = this.configureAndRegisterInvoker(invokerDef, ref, method, parserContext);
String period = element.getAttribute(PERIOD_ATTRIBUTE);
if (StringUtils.hasText(period)) {
adapterDef.getPropertyValues().addPropertyValue("period", period);
if (!StringUtils.hasText(period)) {
throw new ConfigurationException("'period' is required");
}
adapterDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel));
PollingSchedule schedule = new PollingSchedule(Integer.valueOf(period));
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName));
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(channel));
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(schedule);
}
else {
adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class);
invokerDef = new RootBeanDefinition(MethodInvokingTarget.class);
String invokerBeanName = this.configureAndRegisterInvoker(invokerDef, ref, method, parserContext);
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName));
}
invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
invokerDef.getPropertyValues().addPropertyValue("method", method);
String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName));
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName));
adapterDef.setSource(parserContext.extractSource(element));
String beanName = element.getAttribute(ID_ATTRIBUTE);
if (!StringUtils.hasText(beanName)) {
@@ -112,4 +114,12 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
return adapterDef;
}
private String configureAndRegisterInvoker(RootBeanDefinition invokerDef, String objectRef, String methodName, ParserContext parserContext) {
invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
invokerDef.getPropertyValues().addPropertyValue("method", methodName);
String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName));
return invokerBeanName;
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -49,8 +49,7 @@ import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
@@ -161,17 +160,15 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
MethodInvokingSource<Object> source = new MethodInvokingSource<Object>();
source.setObject(bean);
source.setMethod(method.getName());
PollingSourceAdapter<Object> adapter = new PollingSourceAdapter<Object>(source);
MessageChannel channel = new SimpleChannel();
adapter.setChannel(channel);
adapter.setPeriod(period);
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, channel);
messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
SynchronousChannel channel = new SynchronousChannel();
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
Subscription subscription = new Subscription(channel, schedule);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, channel);
messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
Subscription subscription = new Subscription(channel);
endpoint.setSubscription(subscription);
}
}

20
spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java → spring-integration-core/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -14,16 +14,24 @@
* limitations under the License.
*/
package org.springframework.integration.scheduling;
package org.springframework.integration.message;
/**
* Callback interface for components that require the
* {@link MessagingTaskScheduler}.
* Interface that provides callback definitions for components that require
* message delivery status notifications.
*
* @author Mark Fisher
*/
public interface MessagingTaskSchedulerAware {
public interface MessageDeliveryAware {
void setMessagingTaskScheduler(MessagingTaskScheduler scheduler);
/**
* Callback method invoked after a message is sent successfully.
*/
void onSend(Message<?> sentMessage);
/**
* Callback method invoked after a message delivery failure.
*/
void onFailure(MessagingException exception);
}

3
spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java
View File

@@ -23,6 +23,9 @@ package org.springframework.integration.message;
*/
public interface PollableSource<T> {
/**
* Retrieve a message from this source or <code>null</code> if no message is available.
*/
Message<T> poll();
}

35
spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* 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.
@@ -23,6 +23,9 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.util.Assert;
@@ -34,14 +37,14 @@ import org.springframework.util.Assert;
*/
public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler {
private final Log logger = LogFactory.getLog(this.getClass());
private final ScheduledExecutorService executor;
private volatile ErrorHandler errorHandler;
private final Set<Runnable> pendingTasks = new CopyOnWriteArraySet<Runnable>();
private volatile boolean starting;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
@@ -67,17 +70,21 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.running || this.starting) {
if (this.running) {
return;
}
this.starting = true;
this.running = true;
for (Runnable task : this.pendingTasks) {
if (logger.isDebugEnabled()) {
logger.debug("scheduling task: " + task);
}
this.schedule(task);
}
this.pendingTasks.clear();
if (logger.isInfoEnabled()) {
logger.info("task scheduler started successfully");
}
}
for (Runnable task : this.pendingTasks) {
this.schedule(task);
}
this.pendingTasks.clear();
this.running = true;
this.starting = false;
}
public void stop() {
@@ -91,7 +98,7 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler
@Override
public ScheduledFuture<?> schedule(Runnable task) {
if (!this.isRunning()) {
if (!this.running) {
this.pendingTasks.add(task);
return null;
}
@@ -118,9 +125,9 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler
private class MessagingTaskRunner implements Runnable {
private Runnable task;
private final Runnable task;
private boolean shouldRepeat;
private volatile boolean shouldRepeat;
public MessagingTaskRunner(Runnable task) {

31
spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java
View File

@@ -28,6 +28,7 @@ import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* @author Mark Fisher
@@ -38,10 +39,9 @@ public class PollingSourceAdapterTests {
public void testPolledSourceSendsToChannel() {
TestSource source = new TestSource("testing", 1);
SimpleChannel channel = new SimpleChannel();
PollingSourceAdapter<String> adapter = new PollingSourceAdapter<String>(source);
adapter.setChannel(channel);
adapter.setPeriod(100);
adapter.start();
PollingSchedule schedule = new PollingSchedule(100);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
adapter.run();
Message<?> message = channel.receive(1000);
assertNotNull("message should not be null", message);
assertEquals("testing.1", message.getPayload());
@@ -51,22 +51,18 @@ public class PollingSourceAdapterTests {
public void testSendTimeout() {
TestSource source = new TestSource("testing", 1);
SimpleChannel channel = new SimpleChannel(1);
PollingSourceAdapter<String> adapter = new PollingSourceAdapter<String>(source);
adapter.setChannel(channel);
adapter.setInitialDelay(10000);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
adapter.setSendTimeout(10);
adapter.start();
adapter.processMessages();
adapter.processMessages();
adapter.stop();
adapter.run();
Message<?> message1 = channel.receive(1000);
assertNotNull("message should not be null", message1);
assertEquals("testing.1", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertNull("second message should be null", message2);
source.resetCounter();
adapter.start();
adapter.processMessages();
adapter.run();
Message<?> message3 = channel.receive(100);
assertNotNull("third message should not be null", message3);
assertEquals("testing.1", message3.getPayload());
@@ -76,12 +72,11 @@ public class PollingSourceAdapterTests {
public void testMultipleMessagesPerPoll() {
TestSource source = new TestSource("testing", 3);
SimpleChannel channel = new SimpleChannel();
PollingSourceAdapter<String> adapter = new PollingSourceAdapter<String>(source);
adapter.setChannel(channel);
adapter.setInitialDelay(10000);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
adapter.setMaxMessagesPerTask(5);
adapter.start();
adapter.processMessages();
adapter.run();
Message<?> message1 = channel.receive(0);
assertNotNull("message should not be null", message1);
assertEquals("testing.1", message1.getPayload());

7
spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml
View File

@@ -19,7 +19,12 @@
<property name="method" value="foo"/>
</bean>
</constructor-arg>
<property name="channel" ref="channel"/>
<constructor-arg ref="channel"/>
<constructor-arg>
<bean class="org.springframework.integration.scheduling.PollingSchedule">
<constructor-arg value="1000"/>
</bean>
</constructor-arg>
</bean>
<bean id="targetAdapter" class="org.springframework.integration.adapter.DefaultTargetAdapter">

4
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -42,6 +42,7 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
/**
@@ -172,8 +173,7 @@ public class MessageBusTests {
public void testErrorChannelWithFailedDispatch() throws InterruptedException {
MessageBus bus = new MessageBus();
CountDownLatch latch = new CountDownLatch(1);
SourceAdapter sourceAdapter = new PollingSourceAdapter<Object>(new FailingSource(latch));
sourceAdapter.setChannel(new SimpleChannel());
SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch), new SimpleChannel(), new PollingSchedule(1000));
bus.registerSourceAdapter("testAdapter", sourceAdapter);
bus.start();
latch.await(1000, TimeUnit.MILLISECONDS);