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 new file mode 100644 index 0000000000..98c88e52cf --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -0,0 +1,197 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.endpoint; + +import java.util.concurrent.ScheduledFuture; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.scheduling.IntervalTrigger; +import org.springframework.integration.scheduling.TaskScheduler; +import org.springframework.integration.scheduling.TaskSchedulerAware; +import org.springframework.integration.scheduling.Trigger; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.DefaultTransactionDefinition; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + */ +public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSchedulerAware, Lifecycle, InitializingBean { + + public static final int MAX_MESSAGES_UNBOUNDED = -1; + + + private volatile Trigger trigger; + + protected volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED; + + private volatile TaskExecutor taskExecutor; + + private volatile PlatformTransactionManager transactionManager; + + private volatile TransactionDefinition transactionDefinition; + + private volatile TransactionTemplate transactionTemplate; + + private volatile TaskScheduler taskScheduler; + + private volatile ScheduledFuture runningTask; + + private volatile boolean initialized; + + private final Object lifecycleMonitor = new Object(); + + + public void setTrigger(Trigger trigger) { + this.trigger = trigger; + } + + /** + * Set the maximum number of messages to receive for each poll. + * A non-positive value indicates that polling should repeat as long + * as non-null messages are being received and successfully sent. + * + *

The default is unbounded. + * + * @see #MAX_MESSAGES_UNBOUNDED + */ + public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { + this.maxMessagesPerPoll = maxMessagesPerPoll; + } + + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + /** + * Specify a transaction manager to use for all polling operations. + * If none is provided, then the operations will occur without any + * transactional behavior (i.e. there is no default transaction manager). + */ + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setTransactionDefinition(TransactionDefinition transactionDefinition) { + this.transactionDefinition = transactionDefinition; + } + + private TransactionTemplate getTransactionTemplate() { + if (!this.initialized) { + this.afterPropertiesSet(); + } + return this.transactionTemplate; + } + + public void afterPropertiesSet() { + synchronized (this.lifecycleMonitor) { + if (this.initialized) { + return; + } + if (this.trigger == null) { + this.trigger = new IntervalTrigger(0); + } + if (this.transactionManager != null) { + if (this.transactionDefinition == null) { + this.transactionDefinition = new DefaultTransactionDefinition(); + } + this.transactionTemplate = new TransactionTemplate( + this.transactionManager, this.transactionDefinition); + } + this.initialized = true; + } + } + + + // Lifecycle implementation + + public boolean isRunning() { + synchronized (this.lifecycleMonitor) { + return this.runningTask != null; + } + } + + public void start() { + synchronized (this.lifecycleMonitor) { + Assert.state(this.taskScheduler != null, + "unable to start polling, no taskScheduler available"); + this.runningTask = this.taskScheduler.schedule(new Poller(), this.trigger); + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.runningTask != null) { + this.runningTask.cancel(true); + } + this.runningTask = null; + } + } + + + protected abstract boolean doPoll(); + + + private class Poller implements Runnable { + + public void run() { + if (taskExecutor != null) { + taskExecutor.execute(new Runnable() { + public void run() { + poll(); + } + }); + } + else { + poll(); + } + } + + private void poll() { + int count = 0; + while (maxMessagesPerPoll < 0 || count < maxMessagesPerPoll) { + if (!innerPoll()) { + break; + } + count++; + } + } + + private boolean innerPoll() { + TransactionTemplate txTemplate = getTransactionTemplate(); + if (txTemplate != null) { + return (Boolean) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return doPoll(); + } + }); + } + return doPoll(); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java deleted file mode 100644 index 23ee867ea1..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * Copyright 2002-2008 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.endpoint; - -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryAware; -import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.MessagingException; -import org.springframework.integration.scheduling.Trigger; -import org.springframework.util.Assert; - -/** - * @author Mark Fisher - */ -public class SourcePoller extends AbstractPoller { - - private final MessageSource source; - - private final MessageChannel channel; - - - public SourcePoller(MessageSource source, MessageChannel channel, Trigger trigger) { - super(trigger); - Assert.notNull(source, "source must not be null"); - Assert.notNull(channel, "channel must not be null"); - this.source = source; - this.channel = channel; - } - - - @Override - protected boolean doPoll() { - Message message = this.source.receive(); - if (message == null) { - return false; - } - try { - boolean sent = this.channel.send(message); - if (sent && this.source instanceof MessageDeliveryAware) { - ((MessageDeliveryAware) this.source).onSend(message); - } - return sent; - } - catch (Exception e) { - if (this.source instanceof MessageDeliveryAware) { - ((MessageDeliveryAware) this.source).onFailure(message, e); - } - throw (e instanceof MessagingException) ? (MessagingException) e - : new MessageDeliveryException(message, "source poller failed to send message to channel", e); - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 765de8826a..ed5791525d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -16,14 +16,13 @@ package org.springframework.integration.endpoint; -import java.util.concurrent.ScheduledFuture; - -import org.springframework.context.Lifecycle; +import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.MessageChannelTemplate; +import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MethodInvokingSource; -import org.springframework.integration.scheduling.TaskScheduler; -import org.springframework.integration.scheduling.Trigger; +import org.springframework.util.Assert; /** * A Channel Adapter implementation for connecting a @@ -32,73 +31,55 @@ import org.springframework.integration.scheduling.Trigger; * * @author Mark Fisher */ -public class SourcePollingChannelAdapter extends AbstractMessageProducingEndpoint implements Lifecycle { +public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements BeanNameAware { + + private volatile String name; private volatile MessageSource source; - private volatile Trigger trigger; + private volatile MessageChannel outputChannel; - private volatile SourcePoller poller; + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); - private volatile ScheduledFuture pollerFuture; - - private volatile int maxMessagesPerPoll = -1; - - private volatile boolean running; - - private final Object lifecycleMonitor = new Object(); + public void setBeanName(String beanName) { + this.name = beanName; + } public void setSource(MessageSource source) { this.source = source; } - public void setTrigger(Trigger trigger) { - this.trigger = trigger; + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; } - public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { - this.maxMessagesPerPoll = maxMessagesPerPoll; - if (this.poller != null) { - this.poller.setMaxMessagesPerPoll(maxMessagesPerPoll); + public void setSendTimeout(long sendTimeout) { + this.channelTemplate.setSendTimeout(sendTimeout); + } + + public void afterPropertiesSet() { + Assert.notNull(this.source, "source must not be null"); + Assert.notNull(this.outputChannel, "outputChannel must not be null"); + super.afterPropertiesSet(); + if (this.maxMessagesPerPoll < 0 && source instanceof MethodInvokingSource) { + // the default is 1 since a MethodInvokingSource might return + // a non-null value every time it is invoked + this.setMaxMessagesPerPoll(1); } } - public final boolean isRunning() { - return this.running; + @Override + protected boolean doPoll() { + Message message = this.source.receive(); + if (message != null) { + return this.channelTemplate.send(message, this.outputChannel); + } + return false; } - public final void start() { - synchronized (this.lifecycleMonitor) { - if (this.running) { - return; - } - this.poller = new SourcePoller(source, this.getOutputChannel(), trigger); - if (maxMessagesPerPoll < 0 && source instanceof MethodInvokingSource) { - // the default is 1 since a MethodInvokingSource might return a non-null value - // every time it is invoked, thus producing an infinite number of messages per poll - maxMessagesPerPoll = 1; - } - this.configureTransactionSettingsForPoller(this.poller); - this.poller.setMaxMessagesPerPoll(maxMessagesPerPoll); - TaskScheduler taskScheduler = this.getTaskScheduler(); - if (taskScheduler != null) { - this.pollerFuture = taskScheduler.schedule(this.poller, this.poller.getTrigger()); - } - this.running = true; - } - } - - public final void stop() { - synchronized (this.lifecycleMonitor) { - if (!this.running) { - return; - } - if (this.pollerFuture != null) { - this.pollerFuture.cancel(true); - } - this.running = false; - } + public String toString() { + return this.name; } }