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

Incomplete: Improve FileSource

Browse Source 
- refactored Backlog to be more robust (still missing concurrent testcases)
   - refactored FileInfo into FileSnapshot (to remove need for maps in backlog)
   - refactored testcases and AbstractDirectorySource accordingly
This commit is contained in:
Iwein Fuld
2008-08-24 13:24:15 +00:00
parent eba716e65b
commit 8559f2f74b
10 changed files with 558 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,12 +17,11 @@
package org.springframework.integration.adapter.file;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
import org.springframework.integration.message.MessageDeliveryAware;
@@ -31,8 +30,8 @@ import org.springframework.integration.message.PollableSource;
import org.springframework.util.Assert;
/**
* Base class for implementing a PollableSource that creates messages
* from files in a directory, either local or remote.
* Base class for implementing a PollableSource that creates messages from files
* in a directory, either local or remote.
*
* @author Marius Bogoevici
* @author Iwein Fuld
@@ -41,21 +40,18 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
public final static String FILE_INFO_PROPERTY = "file.info";
protected final Log logger = LogFactory.getLog(this.getClass());
private final Backlog<FileInfo> directoryContentManager = new Backlog<FileInfo>();
private final Backlog<FileSnapshot> directoryContentManager = new Backlog<FileSnapshot>();
private final MessageCreator<T, T> messageCreator;
public AbstractDirectorySource(MessageCreator<T, T> messageCreator) {
Assert.notNull(messageCreator, "The MessageCreator must not be null");
this.messageCreator = messageCreator;
}
protected Backlog<FileInfo> getDirectoryContentManager() {
protected Backlog<FileSnapshot> getBacklog() {
return this.directoryContentManager;
}
@@ -63,12 +59,10 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
return this.messageCreator;
}
@SuppressWarnings("unchecked")
public final Message<T> receive() {
try {
refreshSnapshotAndMarkProcessing(this.directoryContentManager);
if (!(getDirectoryContentManager().getProcessingBuffer().isEmpty() & getDirectoryContentManager()
.getBacklog().isEmpty())) {
if (!getBacklog().isEmpty()) {
return buildNextMessage();
}
return null;
@@ -80,15 +74,16 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
/**
* Naive implementation that ignores thread safety. Subclasses that want to
* be thread safe and use the reservation facilities of
* {@link Backlog} should override this method and call
* <code>directoryContentManager.fileProcessing(...)</code> with the appropriate arguments.
* be thread safe and use the reservation facilities of {@link Backlog}
* should override this method and call
* <code>directoryContentManager.fileProcessing(...)</code> with the
* appropriate arguments.
*
* @param directoryContentManager
* @throws IOException
*/
protected void refreshSnapshotAndMarkProcessing(Backlog<FileInfo> directoryContentManager) throws IOException {
HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
protected void refreshSnapshotAndMarkProcessing(Backlog<FileSnapshot> directoryContentManager) throws IOException {
List<FileSnapshot> snapshot = new ArrayList<FileSnapshot>();
this.populateSnapshot(snapshot);
directoryContentManager.processSnapshot(snapshot);
}
@@ -108,12 +103,17 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
return this.messageCreator.createMessage(retrieveNextPayload());
}
public abstract void onSend(Message<?> message);
public void onSend(Message<?> message) {
if (logger.isDebugEnabled()) {
logger.debug(message + " processed successfully. Files will be removed from backlog");
}
this.directoryContentManager.processed();
}
public void onFailure(Message<?> failedMessage, Throwable exception) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("Failure notification received by [" + this.getClass().getSimpleName()
+ "] for message: " + failedMessage, exception);
this.logger.warn("Failure notification received by [" + this.getClass().getSimpleName() + "] for message: "
+ failedMessage + ". Selected files will be moved back to the backlog.", exception);
}
this.directoryContentManager.processingFailed();
}
@@ -124,7 +124,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
* @param snapshot
* @throws IOException
*/
protected abstract void populateSnapshot(Map<String, FileInfo> snapshot) throws IOException;
protected abstract void populateSnapshot(List<FileSnapshot> snapshot) throws IOException;
/**
* Returns the next file, based on the backlog data.
@@ -134,8 +134,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
*/
protected abstract T retrieveNextPayload() throws IOException;
protected final void fileProcessed(String ... fileNames) {
this.directoryContentManager.fileProcessed(fileNames);
protected final void filesProcessed() {
this.directoryContentManager.processed();
}
}

162
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java
View File

@@ -16,44 +16,47 @@
package org.springframework.integration.adapter.file;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.PriorityBlockingQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;
/**
* Tracks changes in a collection. This implementation is thread-safe as it
* allows to synchronously process a new directory structure.
* Keeps track of a backlog in a threadsafe, stateful manner.
*
* @author Marius Bogoevici
* @author Mark Fisher
* @author Iwein Fuld
*/
public class Backlog<T> {
public class Backlog<T extends Comparable<T>> {
private final Log logger = LogFactory.getLog(this.getClass());
private Map<String, T> previousSnapshot = new HashMap<String, T>();
private final Map<String, T> backlog = new HashMap<String, T>();
private PriorityBlockingQueue<T> backlog = new PriorityBlockingQueue<T>();
/**
* This is the storage for backlog that is being processed by a specific
* thread. Not initialized means that we're not in thread safe mode (just
* working directly on the backlog)
*/
private ThreadLocal<Map<String, T>> processingBuffer = new ThreadLocal<Map<String, T>>() {
private ThreadLocal<List<T>> processingBuffer = new ThreadLocal<List<T>>() {
@Override
protected Map<String, T> initialValue() {
return new HashMap<String, T>();
protected List<T> initialValue() {
return new ArrayList<T>();
}
};
public synchronized void processSnapshot(Map<String, T> currentSnapshot) {
private Set<T> doneProcessing = Collections.synchronizedSet(new HashSet<T>());
private Set<T> currentlyProcessing = Collections.synchronizedSet(new HashSet<T>());
public synchronized void processSnapshot(List<T> currentSnapshot) {
/*
* clear the threadLocal backlog. When the thread processes a new
* snapshot it is done with the previous message. If there are still
@@ -61,72 +64,105 @@ public class Backlog<T> {
* were not processed, nor raised as failed.
*/
Assert.isTrue(processingBuffer.get().isEmpty(), "Processing buffer not emptied before poll.");
Iterator<Map.Entry<String, T>> iter = this.backlog.entrySet().iterator();
while (iter.hasNext()) {
String key = iter.next().getKey();
if (!currentSnapshot.containsKey(key)) {
if (logger.isDebugEnabled()) {
logger.debug("Removing item '" + key
+ "' from backlog. It no longer exists in remote directory.");
}
iter.remove();
}
}
for (String key : currentSnapshot.keySet()) {
if (!this.previousSnapshot.containsKey(key)
|| (!this.previousSnapshot.get(key).equals(currentSnapshot.get(key)))) {
if (logger.isDebugEnabled()) {
logger.debug("Adding new or modified item '" + key + "' to backlog.");
}
this.backlog.put(key, currentSnapshot.get(key));
}
}
this.previousSnapshot = new HashMap<String, T>(currentSnapshot);
// remove everything that is not in the latest snapshot from the backlog
backlog.retainAll(currentSnapshot);
doneProcessing.retainAll(currentSnapshot);
// add everything that is new to the backlog preventing side effect on
// currentSnapshot
Collection<T> newInCurrentSnapshot = new ArrayList<T>(currentSnapshot);
newInCurrentSnapshot.removeAll(backlog);
newInCurrentSnapshot.removeAll(doneProcessing);
newInCurrentSnapshot.removeAll(currentlyProcessing);
backlog.addAll(newInCurrentSnapshot);
}
public synchronized void itemProcessing(String... keys) {
for (String key : keys) {
if (key != null) {
/**
* When a source is using a backlog in a single threaded context it can use
* this method instead of using
* <code>{@link #prepareForProcessing(int)}</code>. In a threaded scenario
* this method can be used to mark a subset of the processing buffer as
* processed. Calling this method in a threaded scenario without using
* <code>{@link #prepareForProcessing(int)}</code> will manipulate the
* backlog directly and allows for race conditions. This is only recommended
* when message duplication is not an issue.
* @param items the items that have been processed
*/
public void fileProcessed(T... items) {
for (T item : items) {
if (item != null) {
if (logger.isDebugEnabled()) {
logger.debug("Moving item '" + key
+ "' from the backlog to thread local backlog. It is being processed.");
logger.debug("Removing item '" + item + "' from the undo buffer. It has been processed.");
}
processingBuffer.get().put(key, this.backlog.remove(key));
this.processingBuffer.get().remove(item);
this.backlog.remove(item);
this.doneProcessing.add(item);
}
}
}
public synchronized void processingFailed() {
/**
* Moves items from the backlog to a thread local processing buffer,
* reserving them for this thread. It is the responsibility of the reserving
* thread to process the items and provide feedback to the backlog through
* {@link #processed()} or {@link #processingFailed()}.
* @param maxBatchSize if -1 prepare the whole backlog.
*/
public void prepareForProcessing(int maxBatchSize) {
List<T> processingBuffer = this.processingBuffer.get();
if (maxBatchSize == -1) {
this.backlog.drainTo(processingBuffer);
}
else {
this.backlog.drainTo(processingBuffer, maxBatchSize);
}
currentlyProcessing.addAll(processingBuffer);
}
/**
* Convenience method that returns the items prepared for processing
* immediately.
* @param maxBatchSize
* @return prepared processing buffer
*/
public List<T> selectForProcessing(int maxBatchSize) {
prepareForProcessing(maxBatchSize);
return getProcessingBuffer();
}
/**
* Marks all from the processing buffer as done. Use only in combination
* with <code>{@link #prepareForProcessing(int)}</code>
*/
public void processed() {
this.doneProcessing.addAll(this.processingBuffer.get());
currentlyProcessing.removeAll(processingBuffer.get());
this.processingBuffer.get().clear();
}
/**
* Marks all from the processing buffer as not done. Use only in combination
* with <code>{@link #prepareForProcessing(int)}</code>
*/
public void processingFailed() {
if (logger.isDebugEnabled()) {
logger.debug("Moving all items from processing buffer to backlog. Processing has failed");
}
Map<String, T> processing = this.processingBuffer.get();
this.backlog.putAll(processing);
List<T> processing = this.processingBuffer.get();
currentlyProcessing.removeAll(processing);
this.backlog.addAll(processing);
processing.clear();
}
public synchronized void fileProcessed(String... keys) {
for (String key : keys) {
if (key != null) {
if (logger.isDebugEnabled()) {
logger.debug("Removing item '" + key + "' from the undo buffer. It has been processed.");
}
/*
* It's not relevant if clients use the thread safe approach or
* not at this point, so we act on both.
*/
this.processingBuffer.get().remove(key);
this.backlog.remove(key);
}
}
public List<T> getProcessingBuffer() {
return Collections.unmodifiableList(this.processingBuffer.get());
}
public Map<String, T> getBacklog() {
return Collections.unmodifiableMap(this.backlog);
}
public Map<String, T> getProcessingBuffer() {
return Collections.unmodifiableMap(this.processingBuffer.get());
/**
* @return <code>true</code> if both the thread local processing buffer and
* the backlog are empty.
*/
public boolean isEmpty() {
return this.backlog.isEmpty() && this.processingBuffer.get().isEmpty();
}
}

68
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileInfo.java
View File

@@ -1,68 +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.adapter.file;
/**
* Information about a file in a directory.
*
* @author Marius Bogoevici
*/
public class FileInfo {
private final String fileName;
private final long modificationTimestamp;
private final long size;
public FileInfo(String fileName, long modificationTimestamp, long size) {
this.fileName = fileName;
this.modificationTimestamp = modificationTimestamp;
this.size = size;
}
public String getFileName() {
return fileName;
}
public long getModificationTimestamp() {
return modificationTimestamp;
}
public long getSize() {
return size;
}
@Override
public boolean equals(Object other) {
if (other == null || !(other instanceof FileInfo)) {
return false;
}
FileInfo otherInfo = (FileInfo) other;
return this.getSize() == otherInfo.getSize()
&& this.getModificationTimestamp() == otherInfo.getModificationTimestamp()
&& this.fileName.equals(otherInfo.getFileName());
}
@Override
public int hashCode() {
return (fileName == null ? 0 : fileName.hashCode()) ^ new Long(modificationTimestamp).hashCode()
^ new Long(size).hashCode();
}
}

101
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSnapshot.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* 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.adapter.file;
import java.io.File;
import org.springframework.util.Assert;
/**
* Information about a file.
*
* The FileSnapshot takes a snapshot of certain mutable properties of a file and
* stores them in an immutable way. This can be useful to determine if files
* have been changed.
*
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class FileSnapshot implements Comparable<FileSnapshot> {
private final File file;
private final long modificationTimestamp;
private final long size;
public FileSnapshot(File file) {
Assert.notNull(file, "Can't take a snapshot of file that is null");
this.file = file;
this.modificationTimestamp = file.lastModified();
this.size = file.length();
}
public FileSnapshot(String fileName, long modificationTimestamp, long size) {
this.modificationTimestamp = modificationTimestamp;
this.size = size;
this.file = new File(fileName);
}
public String getFileName() {
// this could be cached for better performance
return file.getName();
}
public long getModificationTimestamp() {
return modificationTimestamp;
}
public long getSize() {
return size;
}
/**
* <p>
* Be careful to note that the file that this snapshot refers to might have
* changed. In particular: <code>
* snapshot.getModificationTimestamp() != snapshot.getFile().lastModified()
* </code> will evalutate to <code>true</code> in
* many scenarios.
*
* @return the file that the snapshot was based on.
*/
public File getFile() {
return file;
}
@Override
public boolean equals(Object other) {
if (other == null || !(other instanceof FileSnapshot)) {
return false;
}
FileSnapshot otherInfo = (FileSnapshot) other;
return this.getSize() == otherInfo.getSize()
&& this.getModificationTimestamp() == otherInfo.getModificationTimestamp()
&& this.file.getName().equals(otherInfo.getFileName());
}
@Override
public int hashCode() {
return file.getPath().hashCode() ^ new Long(modificationTimestamp).hashCode() ^ new Long(size).hashCode();
}
public int compareTo(FileSnapshot other) {
return this.getFile().compareTo(other.getFile());
}
}

61
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java
View File

@@ -20,10 +20,7 @@ import java.io.File;
import java.io.FileFilter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.List;
import org.springframework.core.io.Resource;
import org.springframework.integration.ConfigurationException;
@@ -33,36 +30,33 @@ import org.springframework.integration.message.MessageCreator;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* A messaging source that polls a directory to retrieve files.
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class FileSource extends AbstractDirectorySource<File> implements MessageDeliveryAware {
private final Log logger = LogFactory.getLog(this.getClass());
private final File directory;
private volatile FileFilter fileFilter;
private volatile FilenameFilter filenameFilter;
public FileSource(Resource directory) {
this(directory, new FileMessageCreator());
}
@Override
protected Message<File> buildNextMessage() throws IOException {
File file = retrieveNextPayload();
Message<File> message = this.getMessageCreator().createMessage(file);
message = MessageBuilder.fromMessage(message)
.setHeader(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()).setHeader(FILE_INFO_PROPERTY,
getDirectoryContentManager().getBacklog().get(file.getName())).build();
getBacklog().getProcessingBuffer().get(0)).build();
return message;
}
@@ -81,17 +75,34 @@ public class FileSource extends AbstractDirectorySource<File> implements Message
}
}
/**
* Sets a FilenameFilter to be used with the
* <code>{@link File#listFiles()}</code> command. Note that either a
* FileFilter or a FilenameFilter is used to filter the list of files.
* Calling this setter overwrites the FileFilter if it was set before.
* @param fileFilter
*/
public void setFileFilter(FileFilter fileFilter) {
Assert.notNull(fileFilter);
this.filenameFilter = null;
this.fileFilter = fileFilter;
}
/**
* Sets a FilenameFilter to be used with the
* <code>{@link File#listFiles()}</code> command. Note that either a
* FileFilter or a FilenameFilter is used to filter the list of files.
* Calling this setter overwrites the FileFilter if it was set before.
* @param filenameFilter
*/
public void setFilenameFilter(FilenameFilter filenameFilter) {
Assert.notNull(filenameFilter);
this.fileFilter = null;
this.filenameFilter = filenameFilter;
}
@Override
protected void populateSnapshot(Map<String, FileInfo> snapshot) throws IOException {
protected void populateSnapshot(List<FileSnapshot> snapshot) throws IOException {
File[] files;
if (this.fileFilter != null) {
files = this.directory.listFiles(this.fileFilter);
@@ -103,29 +114,23 @@ public class FileSource extends AbstractDirectorySource<File> implements Message
files = this.directory.listFiles();
}
if (files == null) {
throw new MessagingException("Problem occurred while polling for files. " +
"Is '" + directory.getAbsolutePath() + "' a directory?");
throw new MessagingException("Problem occurred while polling for files. " + "Is '"
+ directory.getAbsolutePath() + "' a directory?");
}
for (File file : files) {
FileInfo fileInfo = new FileInfo(file.getName(), file.lastModified(), file.length());
snapshot.put(file.getName(), fileInfo);
FileSnapshot fileInfo = new FileSnapshot(file);
snapshot.add(fileInfo);
}
}
@Override
protected File retrieveNextPayload() throws IOException {
String fileName = this.getDirectoryContentManager().getBacklog().keySet().iterator().next();
return new File(directory, fileName);
List<FileSnapshot> selectedForProcessing = this.getBacklog().selectForProcessing(1);
if(!selectedForProcessing.isEmpty()){
return selectedForProcessing.get(0).getFile();
} else{
return null;
}
}
@Override
public void onSend(Message<?> message) {
String filename = message.getHeaders().get(FileNameGenerator.FILENAME_PROPERTY_KEY, String.class);
if (StringUtils.hasText(filename)) {
fileProcessed(filename);
}
else if (this.logger.isWarnEnabled()) {
logger.warn("No filename in Message header, cannot send notification of processing.");
}
}
}

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

@@ -20,18 +20,13 @@ import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
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.adapter.file.FileSnapshot;
import org.springframework.integration.message.MessageCreator;
import org.springframework.util.Assert;
@@ -50,13 +45,11 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
private final FTPClientPool clientPool;
public FtpSource(MessageCreator<List<File>, List<File>> messageCreator, FTPClientPool clientPool) {
super(messageCreator);
this.clientPool = clientPool;
}
public void setMaxFilesPerMessage(int maxFilesPerMessage) {
Assert.isTrue(maxFilesPerMessage > 0, "'maxFilesPerMessage' must be greater than 0");
this.maxFilesPerMessage = maxFilesPerMessage;
@@ -68,19 +61,17 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
}
@Override
protected void refreshSnapshotAndMarkProcessing(Backlog<FileInfo> directoryContentManager) throws IOException {
Map<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
protected void refreshSnapshotAndMarkProcessing(Backlog<FileSnapshot> directoryContentManager) throws IOException {
List<FileSnapshot> snapshot = new ArrayList<FileSnapshot>();
synchronized (directoryContentManager) {
populateSnapshot(snapshot);
directoryContentManager.processSnapshot(snapshot);
ArrayList<String> backlog = new ArrayList<String>(directoryContentManager.getBacklog().keySet());
int toIndex = this.maxFilesPerMessage == -1 ? backlog.size() : Math.min(this.maxFilesPerMessage, backlog.size());
directoryContentManager.itemProcessing(backlog.subList(0, toIndex).toArray(new String[] {}));
directoryContentManager.prepareForProcessing(maxFilesPerMessage);
}
}
@Override
protected void populateSnapshot(Map<String, FileInfo> snapshot) throws IOException {
protected void populateSnapshot(List<FileSnapshot> snapshot) throws IOException {
FTPClient client = this.clientPool.getClient();
FTPFile[] fileList = client.listFiles();
try {
@@ -90,9 +81,9 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
* if files couldn't be parsed
*/
if (ftpFile != null) {
FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(),
ftpFile.getSize());
snapshot.put(ftpFile.getName(), fileInfo);
FileSnapshot fileSnapshot = new FileSnapshot(ftpFile.getName(), ftpFile.getTimestamp()
.getTimeInMillis(), ftpFile.getSize());
snapshot.add(fileSnapshot);
}
}
}
@@ -105,14 +96,15 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
FTPClient client = this.clientPool.getClient();
try {
List<File> files = new ArrayList<File>();
Set<String> toDo = this.getDirectoryContentManager().getProcessingBuffer().keySet();
for (String fileName : toDo) {
File file = new File(this.localWorkingDirectory, fileName);
List<FileSnapshot> toDo = this.getBacklog().getProcessingBuffer();
for (FileSnapshot fileSnapshot : toDo) {
//some awkwardness here because the local path may be different from the remote path
File file = new File(this.localWorkingDirectory, fileSnapshot.getFileName());
if (file.exists()) {
file.delete();
}
FileOutputStream fileOutputStream = new FileOutputStream(file);
client.retrieveFile(fileName, fileOutputStream);
client.retrieveFile(fileSnapshot.getFileName(), fileOutputStream);
fileOutputStream.close();
files.add(file);
}
@@ -123,25 +115,4 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
}
}
@Override
public void onSend(Message<?> message) {
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() + "].");
}
}
}

167
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/BacklogTests.java Normal file
View File

@@ -0,0 +1,167 @@
/*
* 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.adapter.file;
import java.util.ArrayList;
import java.util.concurrent.PriorityBlockingQueue;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.adapter.file.Backlog;
import org.springframework.integration.adapter.file.FileSnapshot;
/**
* @author Marius Bogoevici
* @author Iwein Fuld
*/
@SuppressWarnings("unchecked")
public class BacklogTests {
private Backlog<FileSnapshot> backlog;
private ArrayList<FileSnapshot> remoteSnapshot;
private FileSnapshot[] process;
@Before
public void setUp() {
backlog = new Backlog<FileSnapshot>();
process = new FileSnapshot[5];
}
@Test
public void testInitialization() {
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) new DirectFieldAccessor(
backlog).getPropertyValue("backlog");
Assert.assertEquals(3, queue.size());
Assert.assertTrue(queue.containsAll(remoteSnapshot));
}
@Test
public void testFullProcessingInOneStep() {
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed( remoteSnapshot.toArray(process));
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.isEmpty());
}
@Test
public void testFullProcessingInTwoSteps() {
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed( remoteSnapshot.subList(0, 2).toArray(process));
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) new DirectFieldAccessor(
backlog).getPropertyValue("backlog");
Assert.assertEquals(1, queue.size());
Assert.assertTrue(queue.contains(remoteSnapshot.get(2)));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, queue.size());
Assert.assertTrue(queue.contains(remoteSnapshot.get(2)));
backlog.fileProcessed(remoteSnapshot.get(2));
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.isEmpty());
}
@Test
public void testOneFileChangedSize() {
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) new DirectFieldAccessor(
backlog).getPropertyValue("backlog");
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed( remoteSnapshot.toArray(process));
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.isEmpty());
remoteSnapshot.remove(2);
FileSnapshot modifiedC = new FileSnapshot("c.txt", 1001, 112);
remoteSnapshot.add(modifiedC);
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, queue.size());
Assert.assertTrue(queue.contains(modifiedC));
}
@Test
public void testOneFileChangedDate() {
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) new DirectFieldAccessor(
backlog).getPropertyValue("backlog");
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed( remoteSnapshot.toArray(process));
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
remoteSnapshot.remove(2);
FileSnapshot modifiedC = new FileSnapshot("c.txt", 1011, 102);
remoteSnapshot.add(modifiedC);
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, queue.size());
Assert.assertTrue(queue.contains(modifiedC));
}
@Test
public void testOneFileAdded() {
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) new DirectFieldAccessor(
backlog).getPropertyValue("backlog");
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed( remoteSnapshot.toArray(process));
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
FileSnapshot newD = new FileSnapshot("d.txt", 1003, 103);
remoteSnapshot.add(newD);
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, queue.size());
Assert.assertTrue(queue.contains(newD));
}
@Test
public void testOneFileRemoved() {
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed( remoteSnapshot.toArray(process));
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
remoteSnapshot.remove(2);
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.isEmpty());
}
@Test
public void testOneFileRemovedBeforeBeingProcessedInTheNextStep() {
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) new DirectFieldAccessor(
backlog).getPropertyValue("backlog");
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(3, queue.size());
remoteSnapshot.remove(2);
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(2, queue.size());
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(2, queue.size());
}
// @Test selectForProcessing success/failure
@Before
public void generateInitialSnapshot() {
this.remoteSnapshot = new ArrayList<FileSnapshot>();
remoteSnapshot.add(new FileSnapshot("a.txt", 1000, 100));
remoteSnapshot.add(new FileSnapshot("b.txt", 1001, 101));
remoteSnapshot.add(new FileSnapshot("c.txt", 1002, 102));
}
}

103
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/FileSourceTests.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* 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.adapter.file;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.core.io.FileSystemResource;
import org.springframework.integration.message.Message;
/**
* @author Iwein Fuld
*/
public class FileSourceTests {
private FileSource fileSource;
private static final String INPUT_DIR = System.getProperty("java.io.tmpdir") + "/"
+ FileSourceTests.class.getCanonicalName();
@BeforeClass
public static void createTmpDir() {
new File(INPUT_DIR).mkdirs();
}
@Before
public void refreshFileSource() {
fileSource = new FileSource(new FileSystemResource(INPUT_DIR));
}
@After
public void cleanOutTmp() {
for (File file : new File(INPUT_DIR).listFiles()) {
file.delete();
}
}
@AfterClass
public static void removeTmp() {
if (!new File(INPUT_DIR).delete()) {
throw new RuntimeException("failed to clean up directory [" + INPUT_DIR + "]");
}
}
@Test
public void testNoFile() {
assertNull("There should be no message on the source", fileSource.receive());
}
@Test
public void closedEmptyFile() throws Exception {
new File(INPUT_DIR + "/test").createNewFile();
assertNotNull("No file received after writing to input directory", fileSource.receive());
}
@Test
public void closedWriter() throws Exception {
BufferedWriter writer = new BufferedWriter(new FileWriter(new File(INPUT_DIR + "/test")));
writer.write("some stuff");
writer.close();
assertNotNull("No file received after writing to input directory", fileSource.receive());
}
@Test
/*
* This test shows the how not to do it (i.e. without a proper external
* trigger)
*/
public void testOpenWriter() throws Exception {
BufferedWriter writer = new BufferedWriter(new FileWriter(new File(INPUT_DIR + "/test")));
writer.write("some stuff");
// don't close the writer yet
Message<File> received = fileSource.receive();
assertNotNull("incomplete file was not received", received);
fileSource.onSend(received);
writer.write("some more stuff");
writer.close();
assertNotNull("something shoulda happened don't you think?", received);
}
}

180
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/BacklogTests.java
View File

@@ -1,180 +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.adapter.ftp;
import java.util.HashMap;
import java.util.Map;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.adapter.file.Backlog;
import org.springframework.integration.adapter.file.FileInfo;
/**
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class BacklogTests {
private Backlog<FileInfo> backlog;
@Before
public void setUp() {
backlog = new Backlog<FileInfo>();
}
@Test
public void testInitialization() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(3, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("a.txt"));
Assert.assertTrue(backlog.getBacklog().containsKey("b.txt"));
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
}
@Test
public void testFullProcessingInOneStep() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed("a.txt");
backlog.fileProcessed("b.txt");
backlog.fileProcessed("c.txt");
Assert.assertTrue(backlog.getBacklog().isEmpty());
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.getBacklog().isEmpty());
}
@Test
public void testFullProcessingInTwoSteps() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed("a.txt");
backlog.fileProcessed("b.txt");
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
backlog.fileProcessed("c.txt");
Assert.assertTrue(backlog.getBacklog().isEmpty());
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.getBacklog().isEmpty());
}
@Test
public void testOneFileChangedSize() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed("a.txt");
backlog.fileProcessed("b.txt");
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
backlog.fileProcessed("c.txt");
Assert.assertTrue(backlog.getBacklog().isEmpty());
backlog.processSnapshot(remoteSnapshot);
remoteSnapshot.put("c.txt", new FileInfo("c.txt", 1001, 112));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
}
@Test
public void testOneFileChangedDate() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed("a.txt");
backlog.fileProcessed("b.txt");
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
backlog.fileProcessed("c.txt");
Assert.assertTrue(backlog.getBacklog().isEmpty());
backlog.processSnapshot(remoteSnapshot);
remoteSnapshot.put("c.txt", new FileInfo("c.txt", 1011, 102));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
}
@Test
public void testOneFileAdded() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed("a.txt");
backlog.fileProcessed("b.txt");
backlog.fileProcessed("c.txt");
Assert.assertTrue(backlog.getBacklog().isEmpty());
backlog.processSnapshot(remoteSnapshot);
remoteSnapshot.put("d.txt", new FileInfo("d.txt", 1003, 103));
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, backlog.getBacklog().size());
Assert.assertTrue(backlog.getBacklog().containsKey("d.txt"));
}
@Test
public void testOneFileRemoved() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
backlog.fileProcessed("a.txt");
backlog.fileProcessed("b.txt");
backlog.fileProcessed("c.txt");
Assert.assertTrue(backlog.getBacklog().isEmpty());
backlog.processSnapshot(remoteSnapshot);
remoteSnapshot.remove("c.txt");
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.getBacklog().isEmpty());
}
@Test
public void testOneFileRemovedBeforeBeingProcessedInTheNextStep() {
Assert.assertTrue(backlog.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
backlog.processSnapshot(remoteSnapshot);
Assert.assertTrue(backlog.getBacklog().containsKey("c.txt"));
remoteSnapshot.remove("c.txt");
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(2, backlog.getBacklog().size());
backlog.processSnapshot(remoteSnapshot);
Assert.assertEquals(2, backlog.getBacklog().size());
}
private static Map<String, FileInfo> generateInitialSnapshot() {
Map<String, FileInfo> remoteSnapshot = new HashMap<String, FileInfo>();
remoteSnapshot.put("a.txt", new FileInfo("a.txt", 1000, 100));
remoteSnapshot.put("b.txt", new FileInfo("b.txt", 1001, 101));
remoteSnapshot.put("c.txt", new FileInfo("c.txt", 1002, 102));
return remoteSnapshot;
}
}

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

@@ -16,7 +16,15 @@
package org.springframework.integration.adapter.ftp;
import static org.easymock.classextension.EasyMock.*;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.getCurrentArguments;
import static org.easymock.EasyMock.isA;
import static org.easymock.classextension.EasyMock.createMock;
import static org.easymock.classextension.EasyMock.createNiceMock;
import static org.easymock.classextension.EasyMock.replay;
import static org.easymock.classextension.EasyMock.reset;
import static org.easymock.classextension.EasyMock.verify;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -37,7 +45,6 @@ import org.easymock.IAnswer;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
@@ -67,7 +74,6 @@ public class FtpSourceTests {
private Long size = 100l;
@Before
public void initializeFtpSource() {
ftpSource = new FtpSource(messageCreator, ftpClientPool);
@@ -78,7 +84,6 @@ public class FtpSourceTests {
reset(globalMocks);
}
@Test
public void retrieveSingleFile() throws Exception {
@@ -192,9 +197,9 @@ public class FtpSourceTests {
new File("test3") })));
}
@Test(timeout=60000)
@Test(timeout = 6000)
public void concurrentPollingSunnyDay() throws Exception {
final CountDownLatch recorded = new CountDownLatch(1);
this.ftpSource.setMaxFilesPerMessage(2);
// first run
FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3", "test4", "test5");
@@ -215,17 +220,21 @@ public class FtpSourceTests {
return new GenericMessage(getCurrentArguments()[0]);
}
}).times(3);
replay(globalMocks);
recorded.countDown();
final CountDownLatch receivesDone = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
new Thread(new Runnable() {
public void run() {
Message<List<File>> recievedFiles = ftpSource.receive();
receivesDone.countDown();
Message<List<File>> recievedFiles = null;
try {
// make sure receive happens after recording
recorded.await();
recievedFiles = ftpSource.receive();
receivesDone.countDown();
//make sure onSend happens after all receives
receivesDone.await();
}
catch (InterruptedException e) {