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

When a PollableSourceAdapter has a SynchronousChannel, it now registers its source with that channel (and does not start any tasks with the scheduler). This is a temporary implementation of this feature likely to be handled differently after continued source refactoring.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-11 04:33:26 +00:00
parent ffb6266d54
commit 64d46f7e67
13 changed files with 47 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java
View File

@@ -20,10 +20,10 @@ import java.io.File;
import java.io.FileFilter;
import java.io.FilenameFilter;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.util.Assert;
/**
@@ -84,6 +84,7 @@ public class FileSourceAdapter extends PollingSourceAdapter<Object> implements P
if (this.fileNameGenerator != null) {
this.mapper.setFileNameGenerator(this.fileNameGenerator);
}
super.initialize();
}
public Message<Object> poll() {

3
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java
View File

@@ -28,7 +28,6 @@ import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.adapter.file.ByteArrayFileMapper;
import org.springframework.integration.adapter.file.FileNameGenerator;
@@ -36,6 +35,7 @@ import org.springframework.integration.adapter.file.TextFileMapper;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -120,6 +120,7 @@ public class FtpSourceAdapter extends PollingSourceAdapter<Object> implements Po
else {
this.mapper = new ByteArrayFileMapper(this.localWorkingDirectory);
}
super.initialize();
}
@Override

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
View File

@@ -19,9 +19,9 @@ package org.springframework.integration.adapter.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.jms.core.JmsTemplate;
/**

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java
View File

@@ -20,10 +20,10 @@ import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
/**
* A pollable source for receiving bytes from an {@link InputStream}.

2
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
View File

@@ -20,8 +20,8 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.util.Assert;

2
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
View File

@@ -38,6 +38,7 @@ public class CharacterStreamSourceAdapterTests {
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
adapter.setChannel(channel);
adapter.setInitialDelay(10000);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
@@ -55,6 +56,7 @@ public class CharacterStreamSourceAdapterTests {
StringReader reader = new StringReader("test");
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader);
adapter.setInitialDelay(10000);
adapter.setChannel(channel);
adapter.setMaxMessagesPerTask(5);
adapter.start();

1
spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java
View File

@@ -23,6 +23,7 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.handler.HandlerMethodInvoker;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.util.MethodValidator;
import org.springframework.util.Assert;

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

@@ -23,8 +23,10 @@ import java.util.concurrent.Executors;
import org.springframework.context.Lifecycle;
import org.springframework.integration.ConfigurationException;
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.PollableSource;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
@@ -50,10 +52,10 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
private volatile int maxMessagesPerTask = 1;
private volatile boolean starting;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
/**
* Create a new adapter for the given source.
@@ -106,28 +108,38 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
if (this.source == null) {
throw new ConfigurationException("source must not be null");
}
if (this.getChannel() instanceof SynchronousChannel) {
((SynchronousChannel) this.getChannel()).setSource(this.source);
}
}
public void start() {
if (this.isRunning() || this.starting) {
return;
}
this.starting = true;
if (!this.isInitialized()) {
this.afterPropertiesSet();
}
if (this.scheduler == null) {
if (logger.isInfoEnabled()) {
logger.info("no task scheduler has been provided, will create one");
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
return;
}
this.scheduler = new SimpleMessagingTaskScheduler(Executors.newSingleThreadScheduledExecutor());
if (!this.isInitialized()) {
this.afterPropertiesSet();
}
if (this.getChannel() instanceof SynchronousChannel) {
if (logger.isInfoEnabled()) {
logger.info("source adapter configured on synchronous channel, not scheduling");
}
this.running = true;
return;
}
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());
this.running = true;
this.starting = false;
}
public void stop() {

8
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
View File

@@ -22,11 +22,11 @@ import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.selector.MessageSelector;
/**
@@ -48,7 +48,7 @@ public class SynchronousChannel extends AbstractMessageChannel {
private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder();
private final PollableSource<?> source;
private volatile PollableSource<?> source;
private final MessageDistributor distributor;
@@ -66,6 +66,10 @@ public class SynchronousChannel extends AbstractMessageChannel {
}
public void setSource(PollableSource<?> source) {
this.source = source;
}
public void addHandler(MessageHandler handler) {
this.distributor.addHandler(handler);
this.handlerCount.incrementAndGet();

4
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java → spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java
View File

@@ -14,9 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
import org.springframework.integration.message.Message;
package org.springframework.integration.message;
/**
* Interface for any external message source that can be polled.

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

@@ -27,6 +27,7 @@ import org.junit.Test;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
/**
* @author Mark Fisher

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

@@ -29,7 +29,6 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.adapter.SourceAdapter;
import org.springframework.integration.channel.DispatcherPolicy;
@@ -41,6 +40,7 @@ import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
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.Subscription;

2
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
View File

@@ -28,9 +28,9 @@ import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
/**