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

Browse Source
This commit is contained in:
Iwein Fuld
2008-08-11 20:32:52 +00:00
parent df0653f8cd
commit a42f02ec2f
6 changed files with 292 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -43,7 +43,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
private final Log logger = LogFactory.getLog(this.getClass());
private final DirectoryContentManager directoryContentManager = new DirectoryContentManager();
private final Backlog<FileInfo> directoryContentManager = new Backlog<FileInfo>();
private final MessageCreator<T, T> messageCreator;
@@ -52,7 +52,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
this.messageCreator = messageCreator;
}
protected DirectoryContentManager getDirectoryContentManager() {
protected Backlog<FileInfo> getDirectoryContentManager() {
return this.directoryContentManager;
}
@@ -78,12 +78,12 @@ 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 DirectoryContentManager} override this method and call
* <code>directoryContentManager.fileProcessing(...)</code with the appropriate arguments
* {@link Backlog} override this method and call
* <code>directoryContentManager.fileProcessing(...)</code> with the appropriate arguments.
* @param directoryContentManager
* @throws IOException
*/
protected void refreshSnapshotAndMarkProcessing(DirectoryContentManager directoryContentManager) throws IOException {
protected void refreshSnapshotAndMarkProcessing(Backlog<FileInfo> directoryContentManager) throws IOException {
HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
this.populateSnapshot(snapshot);
directoryContentManager.processSnapshot(snapshot);

69
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java → org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java
View File

@@ -20,41 +20,40 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;
/**
* Tracks changes in a directory. This implementation is thread-safe as it
* Tracks changes in a collection. This implementation is thread-safe as it
* allows to synchronously process a new directory structure.
*
* @author Marius Bogoevici
* @author Mark Fisher
* @author iwein
* @author Iwein Fuld
*/
public class DirectoryContentManager {
public class Backlog<T> {
private final Log logger = LogFactory.getLog(this.getClass());
private Map<String, FileInfo> previousSnapshot = new HashMap<String, FileInfo>();
private Map<String, T> previousSnapshot = new HashMap<String, T>();
private final Map<String, FileInfo> backlog = new HashMap<String, FileInfo>();
private final Map<String, T> backlog = new HashMap<String, 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, FileInfo>> processingBuffer = new ThreadLocal<Map<String, FileInfo>>() {
private ThreadLocal<Map<String, T>> processingBuffer = new ThreadLocal<Map<String, T>>() {
@Override
protected Map<String, FileInfo> initialValue() {
return new HashMap<String, FileInfo>();
protected Map<String, T> initialValue() {
return new HashMap<String, T>();
}
};
public synchronized void processSnapshot(Map<String, FileInfo> currentSnapshot) {
public synchronized void processSnapshot(Map<String, T> currentSnapshot) {
/*
* clear the threadLocal backlog. When the thread processes a new
* snapshot it is done with the previous message. If there are still
@@ -62,71 +61,71 @@ public class DirectoryContentManager {
* were not processed, nor raised as failed.
*/
Assert.isTrue(processingBuffer.get().isEmpty(), "Processing buffer not emptied before poll.");
Iterator<Map.Entry<String, FileInfo>> iter = this.backlog.entrySet().iterator();
Iterator<Map.Entry<String, T>> iter = this.backlog.entrySet().iterator();
while (iter.hasNext()) {
String fileName = iter.next().getKey();
if (!currentSnapshot.containsKey(fileName)) {
String key = iter.next().getKey();
if (!currentSnapshot.containsKey(key)) {
if (logger.isDebugEnabled()) {
logger.debug("Removing file '" + fileName
logger.debug("Removing item '" + key
+ "' from backlog. It no longer exists in remote directory.");
}
iter.remove();
}
}
for (String fileName : currentSnapshot.keySet()) {
if (!this.previousSnapshot.containsKey(fileName)
|| (!this.previousSnapshot.get(fileName).equals(currentSnapshot.get(fileName)))) {
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 file '" + fileName + "' to backlog.");
logger.debug("Adding new or modified item '" + key + "' to backlog.");
}
this.backlog.put(fileName, currentSnapshot.get(fileName));
this.backlog.put(key, currentSnapshot.get(key));
}
}
this.previousSnapshot = new HashMap<String, FileInfo>(currentSnapshot);
this.previousSnapshot = new HashMap<String, T>(currentSnapshot);
}
public synchronized void fileProcessing(String... fileNames) {
for (String fileName : fileNames) {
if (fileName != null) {
public synchronized void itemProcessing(String... keys) {
for (String key : keys) {
if (key != null) {
if (logger.isDebugEnabled()) {
logger.debug("Moving file '" + fileName
logger.debug("Moving item '" + key
+ "' from the backlog to thread local backlog. It is being processed.");
}
processingBuffer.get().put(fileName, this.backlog.remove(fileName));
processingBuffer.get().put(key, this.backlog.remove(key));
}
}
}
public synchronized void processingFailed() {
if (logger.isDebugEnabled()) {
logger.debug("Moving all files from processing buffer to backlog. Processing has failed");
logger.debug("Moving all items from processing buffer to backlog. Processing has failed");
}
Map<String, FileInfo> processing = this.processingBuffer.get();
Map<String, T> processing = this.processingBuffer.get();
this.backlog.putAll(processing);
processing.clear();
}
public synchronized void fileProcessed(String... fileNames) {
for (String fileName : fileNames) {
if (fileName != null) {
public synchronized void fileProcessed(String... keys) {
for (String key : keys) {
if (key != null) {
if (logger.isDebugEnabled()) {
logger.debug("Removing file '" + fileName + "' from the undo buffer. It has been processed.");
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(fileName);
this.backlog.remove(fileName);
this.processingBuffer.get().remove(key);
this.backlog.remove(key);
}
}
}
public Map<String, FileInfo> getBacklog() {
public Map<String, T> getBacklog() {
return Collections.unmodifiableMap(this.backlog);
}
public Map<String, FileInfo> getProcessingBuffer() {
public Map<String, T> getProcessingBuffer() {
return Collections.unmodifiableMap(this.processingBuffer.get());
}

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

@@ -33,7 +33,7 @@ import org.apache.commons.net.ftp.FTPFile;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.integration.adapter.file.AbstractDirectorySource;
import org.springframework.integration.adapter.file.DirectoryContentManager;
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;
@@ -117,14 +117,14 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
}
@Override
protected void refreshSnapshotAndMarkProcessing(DirectoryContentManager directoryContentManager) throws IOException {
protected void refreshSnapshotAndMarkProcessing(Backlog<FileInfo> directoryContentManager) throws IOException {
Map<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
synchronized (directoryContentManager) {
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());
directoryContentManager.fileProcessing(backlog.subList(0, toIndex).toArray(new String[] {}));
directoryContentManager.itemProcessing(backlog.subList(0, toIndex).toArray(new String[] {}));
}
}
@@ -220,5 +220,4 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
fileProcessed(file.getName());
}
}
}