Removed SourcePoller. SourcePollingChannelAdapter now extends AbstractPollingEndpoint.
This commit is contained in:
Mark Fisher
@@ -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.
|
||||
*
|
||||
* <p>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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user