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

tested and fixed onFailure for FtpSource

Browse Source
This commit is contained in:
Iwein Fuld
2008-08-08 14:56:51 +00:00
parent fbb5123e8e
commit e88776463a
7 changed files with 105 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -110,6 +110,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
if (this.logger.isWarnEnabled()) {
logger.warn("Failure notification received by " + this.getClass().getSimpleName(), exception);
}
directoryContentManager.processingFailed();
}
/**
@@ -126,8 +127,8 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
*/
protected abstract T retrieveNextPayload() throws IOException;
protected final void fileProcessed(String fileName) {
this.directoryContentManager.fileProcessed(fileName);
protected final void fileProcessed(String ... fileNames) {
this.directoryContentManager.fileProcessed(fileNames);
}
}

16
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java
View File

@@ -61,7 +61,7 @@ public class DirectoryContentManager {
* messages in the processing buffer something is wrong because they
* were not processed, nor raised as failed.
*/
Assert.isTrue(processingBuffer.get().isEmpty());
Assert.isTrue(processingBuffer.get().isEmpty(), "Processing buffer not emptied before poll.");
Iterator<Map.Entry<String, FileInfo>> iter = this.backlog.entrySet().iterator();
while (iter.hasNext()) {
String fileName = iter.next().getKey();
@@ -97,15 +97,13 @@ public class DirectoryContentManager {
}
}
public synchronized void fileNotProcessed(String... fileNames) {
for (String fileName : fileNames) {
if (fileName != null) {
if (logger.isDebugEnabled()) {
logger.debug("File '" + fileName + "' was not processed. Moving it back to the backlog");
}
this.backlog.put(fileName, this.processingBuffer.get().remove(fileName));
}
public synchronized void processingFailed() {
if (logger.isDebugEnabled()) {
logger.debug("Moving all files from processing buffer to backlog. Processing has failed");
}
Map<String, FileInfo> processing = this.processingBuffer.get();
this.backlog.putAll(processing);
processing.clear();
}
public synchronized void fileProcessed(String... fileNames) {