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

Renamed 'setMaxMessagesPerPayload()' method to 'setMaxFilesPerMessage()'.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-21 13:15:46 +00:00
parent 42c60752c2
commit 3b256d9e81
3 changed files with 34 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java
View File

@@ -41,7 +41,8 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
public final static String FILE_INFO_PROPERTY = "file.info";
private final Log logger = LogFactory.getLog(this.getClass());
protected final Log logger = LogFactory.getLog(this.getClass());
private final Backlog<FileInfo> directoryContentManager = new Backlog<FileInfo>();

51
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
View File

@@ -25,20 +25,15 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.integration.adapter.file.AbstractDirectorySource;
import org.springframework.integration.adapter.file.Backlog;
import org.springframework.integration.adapter.file.FileInfo;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
import org.springframework.integration.message.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* A source adapter for receiving files via FTP.
@@ -49,22 +44,22 @@ import org.springframework.util.StringUtils;
*/
public class FtpSource extends AbstractDirectorySource<List<File>> {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile File localWorkingDirectory;
private int maxFilesPerPayload = -1;
private volatile int maxFilesPerMessage = -1;
private final FTPClientPool clientPool;
public FtpSource(MessageCreator<List<File>, List<File>> messageCreator, FTPClientPool clientPool) {
super(messageCreator);
this.clientPool = clientPool;
}
public void setMaxMessagesPerPayload(int maxMessagesPerPayload) {
Assert.isTrue(maxMessagesPerPayload > 0, "'maxMessagesPerPayload' should greater than 0");
this.maxFilesPerPayload = maxMessagesPerPayload;
public void setMaxFilesPerMessage(int maxFilesPerMessage) {
Assert.isTrue(maxFilesPerMessage > 0, "'maxFilesPerMessage' must be greater than 0");
this.maxFilesPerMessage = maxFilesPerMessage;
}
public void setLocalWorkingDirectory(File localWorkingDirectory) {
@@ -79,14 +74,14 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
populateSnapshot(snapshot);
directoryContentManager.processSnapshot(snapshot);
ArrayList<String> backlog = new ArrayList<String>(directoryContentManager.getBacklog().keySet());
int toIndex = maxFilesPerPayload == -1 ? backlog.size() : Math.min(maxFilesPerPayload, backlog.size());
int toIndex = this.maxFilesPerMessage == -1 ? backlog.size() : Math.min(this.maxFilesPerMessage, backlog.size());
directoryContentManager.itemProcessing(backlog.subList(0, toIndex).toArray(new String[] {}));
}
}
@Override
protected void populateSnapshot(Map<String, FileInfo> snapshot) throws IOException {
FTPClient client = clientPool.getClient();
FTPClient client = this.clientPool.getClient();
FTPFile[] fileList = client.listFiles();
try {
for (FTPFile ftpFile : fileList) {
@@ -102,12 +97,12 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
}
}
finally {
clientPool.releaseClient(client);
this.clientPool.releaseClient(client);
}
}
protected List<File> retrieveNextPayload() throws IOException {
FTPClient client = clientPool.getClient();
FTPClient client = this.clientPool.getClient();
try {
List<File> files = new ArrayList<File>();
Set<String> toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet();
@@ -124,17 +119,29 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
return files;
}
finally {
clientPool.releaseClient(client);
this.clientPool.releaseClient(client);
}
}
@SuppressWarnings("unchecked")
@Override
public void onSend(Message<?> message) {
List<File> files = ((Message<List<File>>) message).getPayload();
for (File file : files) {
fileProcessed(file.getName());
Object payload = message.getPayload();
if (payload instanceof List) {
List<?> items = (List<?>) payload;
for (Object item : items) {
if (item instanceof File) {
fileProcessed(((File) item).getName());
}
else if (logger.isWarnEnabled()) {
logger.warn("FtpSource.onSend() expects Files in the List payload, "
+ "but received an item of type [" + item.getClass() + "]");
}
}
}
else if (logger.isWarnEnabled()) {
logger.warn("FtpSource.onSend() exepects a Message with a List of Files, "
+ "but received payload of type [" + message.getPayload().getClass() + "].");
}
}
}

6
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java
View File

@@ -156,9 +156,9 @@ public class FtpSourceTests {
}
@Test
public void retrieveMaxFilesPerPayload() throws Exception {
public void retrieveMaxFilesPerMessage() throws Exception {
this.ftpSource.setMaxMessagesPerPayload(2);
this.ftpSource.setMaxFilesPerMessage(2);
// assume client already connected
FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3");
@@ -195,7 +195,7 @@ public class FtpSourceTests {
@Test(timeout=60000)
public void concurrentPollingSunnyDay() throws Exception {
this.ftpSource.setMaxMessagesPerPayload(2);
this.ftpSource.setMaxFilesPerMessage(2);
// first run
FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3", "test4", "test5");
expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);