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

PollableSource no longer accepts a "limit" argument in its poll() method (INT-181).

Browse Source
This commit is contained in:
Mark Fisher
2008-04-09 16:33:16 +00:00
parent fec6dec258
commit 6e30173509
14 changed files with 123 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java
View File

@@ -19,9 +19,6 @@ package org.springframework.integration.adapter.file;
import java.io.File;
import java.io.FileFilter;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessagingException;
@@ -54,13 +51,13 @@ public class FileSource implements PollableSource<File> {
this.filenameFilter = filenameFilter;
}
public Collection<File> poll(int limit) {
public File poll() {
File[] files = null;
if (this.fileFilter != null) {
files = this.directory.listFiles(fileFilter);
files = this.directory.listFiles(this.fileFilter);
}
else if (this.filenameFilter != null) {
files = this.directory.listFiles(filenameFilter);
files = this.directory.listFiles(this.filenameFilter);
}
else {
files = this.directory.listFiles();
@@ -69,14 +66,12 @@ public class FileSource implements PollableSource<File> {
throw new MessagingException("Problem occurred while polling for files. " +
"Is '" + directory.getAbsolutePath() + "' a directory?");
}
List<File> results = new ArrayList<File>();
int size = Math.min(limit, files.length);
for (int i = 0; i < size; i++) {
for (int i = 0; i < files.length; i++) {
if (files[i].isFile()) {
results.add(files[i]);
return files[i];
}
}
return results;
return null;
}
}

18
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java
View File

@@ -22,7 +22,7 @@ import java.util.Iterator;
import java.util.Map;
/**
* Tracks changes in the context. This implementation is thread-safe as it
* Tracks changes in a directory. This implementation is thread-safe as it
* allows to synchronously process a new directory structure.
*
* @author Marius Bogoevici
@@ -30,26 +30,26 @@ import java.util.Map;
*/
public class DirectoryContentManager {
private Map<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
private Map<String, FileInfo> previousSnapshot = new HashMap<String, FileInfo>();
private final Map<String, FileInfo> backlog = new HashMap<String, FileInfo>();
public synchronized void processSnapshot(Map<String, FileInfo> remoteSnapshot) {
public synchronized void processSnapshot(Map<String, FileInfo> currentSnapshot) {
Iterator<Map.Entry<String, FileInfo>> iter = this.backlog.entrySet().iterator();
while (iter.hasNext()) {
String fileName = iter.next().getKey();
if (!remoteSnapshot.containsKey(fileName)) {
if (!currentSnapshot.containsKey(fileName)) {
iter.remove();
}
}
for (String fileName : remoteSnapshot.keySet()) {
if (!this.snapshot.containsKey(fileName)
|| (!this.snapshot.get(fileName).equals(remoteSnapshot.get(fileName)))) {
this.backlog.put(fileName, remoteSnapshot.get(fileName));
for (String fileName : currentSnapshot.keySet()) {
if (!this.previousSnapshot.containsKey(fileName)
|| (!this.previousSnapshot.get(fileName).equals(currentSnapshot.get(fileName)))) {
this.backlog.put(fileName, currentSnapshot.get(fileName));
}
}
this.snapshot = new HashMap<String, FileInfo>(remoteSnapshot);
this.previousSnapshot = new HashMap<String, FileInfo>(currentSnapshot);
}
public synchronized void fileProcessed(String fileName) {

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

@@ -19,9 +19,8 @@ package org.springframework.integration.adapter.ftp;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -124,50 +123,30 @@ public class FtpSourceAdapter extends PollingSourceAdapter<File> implements Poll
}
public final Collection<File> poll(int limit) {
public final File poll() {
try {
LinkedList<File> localFileList = new LinkedList<File>();
this.client.connect(this.host, this.port);
if (!StringUtils.hasText(this.username)) {
throw new MessagingException("username is required");
}
if (!this.client.login(this.username, this.password)) {
throw new MessagingException("Login failed. Please check the username and password.");
}
if (logger.isDebugEnabled()) {
logger.debug("login successful");
}
this.client.setFileType(FTP.IMAGE_FILE_TYPE);
if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory())
&& !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) {
throw new MessagingException("Could not change directory to '" +
remoteWorkingDirectory + "'. Please check the path.");
}
if (logger.isDebugEnabled()) {
logger.debug("working directory is: " + this.client.printWorkingDirectory());
}
this.establishConnection();
FTPFile[] fileList = this.client.listFiles();
HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
for (FTPFile ftpFile : fileList) {
FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(),
ftpFile.getSize());
FileInfo fileInfo = new FileInfo(
ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile.getSize());
snapshot.put(ftpFile.getName(), fileInfo);
}
this.directoryContentManager.processSnapshot(snapshot);
for (String fileName : this.directoryContentManager.getBacklog().keySet()) {
File file = new File(this.localWorkingDirectory, fileName);
if (file.exists()) {
file.delete();
}
FileOutputStream fileOutputStream = new FileOutputStream(file);
this.client.retrieveFile(fileName, fileOutputStream);
fileOutputStream.close();
localFileList.add(file);
if (limit >= localFileList.size()) {
break;
}
Map<String, FileInfo> backlog = this.directoryContentManager.getBacklog();
if (backlog.isEmpty()) {
return null;
}
return localFileList;
String fileName = backlog.keySet().iterator().next();
File file = new File(this.localWorkingDirectory, fileName);
if (file.exists()) {
file.delete();
}
FileOutputStream fileOutputStream = new FileOutputStream(file);
this.client.retrieveFile(fileName, fileOutputStream);
fileOutputStream.close();
return file;
}
catch (Exception e) {
try {
@@ -182,4 +161,26 @@ public class FtpSourceAdapter extends PollingSourceAdapter<File> implements Poll
}
}
private void establishConnection() throws IOException {
if (!StringUtils.hasText(this.username)) {
throw new MessagingException("username is required");
}
this.client.connect(this.host, this.port);
if (!this.client.login(this.username, this.password)) {
throw new MessagingException("Login failed. Please check the username and password.");
}
if (logger.isDebugEnabled()) {
logger.debug("login successful");
}
this.client.setFileType(FTP.IMAGE_FILE_TYPE);
if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory())
&& !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) {
throw new MessagingException("Could not change directory to '" +
remoteWorkingDirectory + "'. Please check the path.");
}
if (logger.isDebugEnabled()) {
logger.debug("working directory is: " + this.client.printWorkingDirectory());
}
}
}

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

@@ -16,9 +16,6 @@
package org.springframework.integration.adapter.jms;
import java.util.Arrays;
import java.util.Collection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
@@ -52,8 +49,8 @@ public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implement
}
public Collection<Object> poll(int limit) {
return Arrays.asList(this.getJmsTemplate().receiveAndConvert());
public Object poll() {
return this.getJmsTemplate().receiveAndConvert();
}
}

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

@@ -19,9 +19,6 @@ package org.springframework.integration.adapter.stream;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessagingException;
@@ -68,36 +65,32 @@ public class ByteStreamSource implements PollableSource<byte[]> {
this.shouldTruncate = shouldTruncate;
}
public Collection<byte[]> poll(int limit) {
List<byte[]> results = new ArrayList<byte[]>();
while (results.size() < limit) {
try {
byte[] bytes;
int bytesRead = 0;
synchronized (this.streamMonitor) {
if (stream.available() == 0) {
return results;
}
bytes = new byte[bytesPerMessage];
bytesRead = stream.read(bytes, 0, bytes.length);
}
if (bytesRead <= 0) {
return results;
}
if (!this.shouldTruncate) {
results.add(bytes);
}
else {
byte[] result = new byte[bytesRead];
System.arraycopy(bytes, 0, result, 0, result.length);
results.add(result);
public byte[] poll() {
try {
byte[] bytes;
int bytesRead = 0;
synchronized (this.streamMonitor) {
if (stream.available() == 0) {
return null;
}
bytes = new byte[bytesPerMessage];
bytesRead = stream.read(bytes, 0, bytes.length);
}
catch (IOException e) {
throw new MessagingException("IO failure occurred in adapter", e);
if (bytesRead <= 0) {
return null;
}
if (!this.shouldTruncate) {
return bytes;
}
else {
byte[] result = new byte[bytesRead];
System.arraycopy(bytes, 0, result, 0, result.length);
return result;
}
}
return results;
catch (IOException e) {
throw new MessagingException("IO failure occurred in adapter", e);
}
}
}

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

@@ -19,9 +19,6 @@ package org.springframework.integration.adapter.stream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessagingException;
@@ -58,28 +55,18 @@ public class CharacterStreamSource implements PollableSource<String> {
}
public Collection<String> poll(int limit) {
List<String> results = new ArrayList<String>();
while (results.size() < limit) {
try {
String line = null;
synchronized (this.monitor) {
boolean isReady = this.reader.ready();
if (!isReady) {
return results;
}
line = this.reader.readLine();
public String poll() {
try {
synchronized (this.monitor) {
if (!this.reader.ready()) {
return null;
}
if (line == null) {
return results;
}
results.add(line);
}
catch (IOException e) {
throw new MessagingException("IO failure occurred in adapter", e);
return this.reader.readLine();
}
}
return results;
catch (IOException e) {
throw new MessagingException("IO failure occurred in adapter", e);
}
}
}

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

@@ -17,8 +17,6 @@
package org.springframework.integration.adapter;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collection;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.ConfigurationException;
@@ -56,11 +54,11 @@ public class MethodInvokingSource<T> implements PollableSource<Object>, Initiali
this.invoker.setMethodValidator(new MessageReceivingMethodValidator());
}
public Collection<Object> poll(int limit) {
public Object poll() {
if (this.invoker == null) {
this.afterPropertiesSet();
}
return Arrays.asList(this.invoker.invokeMethod(new Object[] {}));
return this.invoker.invokeMethod(new Object[] {});
}

6
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.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 java.util.Collection;
/**
* Interface for any external data source that can be polled.
*
@@ -25,6 +23,6 @@ import java.util.Collection;
*/
public interface PollableSource<T> {
Collection<T> poll(int limit);
T poll();
}

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

@@ -16,14 +16,12 @@
package org.springframework.integration.adapter;
import java.util.Collection;
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.message.MessageMapper;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
@@ -142,16 +140,14 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
}
int messagesProcessed = 0;
int limit = this.maxMessagesPerTask;
Collection<T> results = this.source.poll(limit);
if (results != null) {
if (results.size() > limit) {
throw new MessagingException("source returned too many results, the limit is " + limit);
while (messagesProcessed < limit) {
T result = this.source.poll();
if (result != null && this.sendToChannel(result)) {
messagesProcessed++;
this.onSend(result);
}
for (T next : results) {
if (this.sendToChannel(next)) {
messagesProcessed++;
this.onSend(next);
}
else {
return messagesProcessed;
}
}
return messagesProcessed;

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

@@ -17,7 +17,6 @@
package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -30,7 +29,6 @@ import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.SimplePayloadMessageMapper;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.util.CollectionUtils;
/**
* A channel that invokes the subscribed {@link MessageHandler handler(s)} in a
@@ -86,9 +84,8 @@ public class SynchronousChannel extends AbstractMessageChannel {
@Override
protected Message<?> doReceive(long timeout) {
if (this.source != null) {
Collection<?> results = this.source.poll(1);
if (!CollectionUtils.isEmpty(results)) {
Object result = results.iterator().next();
Object result = this.source.poll();
if (result != null) {
return (result instanceof Message<?>) ? (Message<?>) result :
new SimplePayloadMessageMapper<Object>().toMessage(result);
}

12
spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java
View File

@@ -19,8 +19,6 @@ package org.springframework.integration.adapter;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.util.Collection;
import org.junit.Test;
import org.springframework.integration.message.MessagingException;
@@ -35,9 +33,9 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("validMethod");
Collection<Object> result = source.poll(5);
Object result = source.poll();
assertNotNull(result);
assertEquals("valid", result.iterator().next());
assertEquals("valid", result);
}
@Test(expected=MessagingException.class)
@@ -45,7 +43,7 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("noMatchingMethod");
source.poll(5);
source.poll();
}
@Test(expected=MessagingException.class)
@@ -53,7 +51,7 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("invalidMethodWithArg");
source.poll(5);
source.poll();
}
@Test(expected=MessagingException.class)
@@ -61,7 +59,7 @@ public class MethodInvokingSourceTests {
MethodInvokingSource<TestBean> source = new MethodInvokingSource<TestBean>();
source.setObject(new TestBean());
source.setMethod("invalidMethodWithNoReturnValue");
source.poll(5);
source.poll();
}

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

@@ -20,16 +20,12 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
/**
* @author Mark Fisher
@@ -66,11 +62,12 @@ public class PollingSourceAdapterTests {
assertEquals("testing.1", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertNull("second message should be null", message2);
source.resetCounter();
adapter.start();
adapter.processMessages();
Message<?> message3 = channel.receive(100);
assertNotNull("third message should not be null", message3);
assertEquals("testing.3", message3.getPayload());
assertEquals("testing.1", message3.getPayload());
}
@Test
@@ -96,38 +93,29 @@ public class PollingSourceAdapterTests {
assertNull("message should be null", message4);
}
@Test(expected=MessagingException.class)
public void testResultSizeExceedsLimit() {
TestSource source = new TestSource("testing", 3);
SimpleChannel channel = new SimpleChannel();
PollingSourceAdapter<String> adapter = new PollingSourceAdapter<String>(source);
adapter.setChannel(channel);
adapter.setPeriod(1000);
adapter.setMaxMessagesPerTask(2);
adapter.start();
adapter.processMessages();
}
private static class TestSource implements PollableSource<String> {
private String message;
private int messagesPerPoll;
private int limit;
private AtomicInteger count = new AtomicInteger();
public TestSource(String message, int messagesPerPoll) {
public TestSource(String message, int limit) {
this.message = message;
this.messagesPerPoll = messagesPerPoll;
this.limit = limit;
}
public Collection<String> poll(int limit) {
List<String> results = new ArrayList<String>(this.messagesPerPoll);
for (int i = 0; i < this.messagesPerPoll; i++) {
results.add(message + "." + count.incrementAndGet());
public void resetCounter() {
this.count.set(0);
}
public String poll() {
if (count.get() >= limit) {
return null;
}
return results;
return message + "." + count.incrementAndGet();
}
}

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

@@ -269,7 +269,7 @@ public class MessageBusTests {
this.latch = latch;
}
public Collection<Object> poll(int limit) {
public Collection<Object> poll() {
latch.countDown();
throw new RuntimeException("intentional test failure");
}

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

@@ -21,8 +21,6 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.SynchronousQueue;
@@ -99,8 +97,8 @@ public class SynchronousChannelTests {
@Test
public void testReceive() {
SynchronousChannel channel = new SynchronousChannel(new PollableSource<String>() {
public Collection<String> poll(int limit) {
return Collections.singleton("foo");
public String poll() {
return "foo";
}
});
Message<?> message = channel.receive();
@@ -186,10 +184,10 @@ public class SynchronousChannelTests {
this.messageText = messageText;
}
public Collection<StringMessage> poll(int limit) {
public StringMessage poll() {
StringMessage message = new StringMessage(messageText);
message.getHeader().setProperty(HANDLER_THREAD, Thread.currentThread().getName());
return Collections.singleton(message);
return message;
}
}