diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java index 5b363a3b82..9403a3c839 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java @@ -35,7 +35,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractPersistentAcceptOnceFileListFilter extends AbstractFileListFilter - implements ReversibleFileListFilter, Closeable { + implements ReversibleFileListFilter, ResettableFileListFilter, Closeable { protected final ConcurrentMetadataStore store; @@ -102,12 +102,18 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter extends Abst rollingBack = true; } if (rollingBack) { - this.store.remove(buildKey(fileToRollback)); - flushIfNeeded(); + remove(fileToRollback); } } } + @Override + public boolean remove(F fileToRemove) { + String removed = this.store.remove(buildKey(fileToRemove)); + flushIfNeeded(); + return removed != null; + } + @Override public void close() throws IOException { if (this.store instanceof Closeable) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AcceptOnceFileListFilter.java index 1456145553..97fcd67979 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AcceptOnceFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AcceptOnceFileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -34,7 +34,8 @@ import java.util.concurrent.LinkedBlockingQueue; * @author Gary Russell * @since 1.0.0 */ -public class AcceptOnceFileListFilter extends AbstractFileListFilter implements ReversibleFileListFilter { +public class AcceptOnceFileListFilter extends AbstractFileListFilter implements ReversibleFileListFilter, + ResettableFileListFilter { private final Queue seen; @@ -93,13 +94,19 @@ public class AcceptOnceFileListFilter extends AbstractFileListFilter imple rollingBack = true; } if (rollingBack) { - this.seenSet.remove(fileToRollback); - if (this.seen != null) { - this.seen.remove(fileToRollback); - } + remove(fileToRollback); } } } } + @Override + public boolean remove(F fileToRemove) { + boolean removed = this.seenSet.remove(fileToRemove); + if (this.seen != null) { + this.seen.remove(fileToRemove); + } + return removed; + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java index 7bfd635b3e..6b392bc32f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java @@ -39,7 +39,7 @@ import org.springframework.util.Assert; * * @param The type that will be filtered. */ -public class CompositeFileListFilter implements FileListFilter, Closeable { +public class CompositeFileListFilter implements ReversibleFileListFilter, Closeable { private final Set> fileFilters; @@ -111,4 +111,13 @@ public class CompositeFileListFilter implements FileListFilter, Closeable return results; } + @Override + public void rollback(F file, List files) { + for (FileListFilter fileFilter : this.fileFilters) { + if (fileFilter instanceof ReversibleFileListFilter) { + ((ReversibleFileListFilter) fileFilter).rollback(file, files); + } + } + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ResettableFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ResettableFileListFilter.java new file mode 100644 index 0000000000..125016b139 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/ResettableFileListFilter.java @@ -0,0 +1,34 @@ +/* + * Copyright 2015 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.file.filters; + +/** + * A {@link FileListFilter} that can be reset by removing a specific file from its + * state. + * @author Gary Russell + * @since 4.1.7 + * + */ +public interface ResettableFileListFilter extends FileListFilter { + + /** + * Remove the specified file from the filter so it will pass on the next attempt. + * @param f the element to remove. + * @return true if the file was removed as a result of this call. + */ + boolean remove(F f); + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/AcceptOnceFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/AcceptOnceFileListFilterTests.java index 2a36e87fd9..5105d96cb1 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/AcceptOnceFileListFilterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/AcceptOnceFileListFilterTests.java @@ -24,6 +24,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Queue; import java.util.Set; @@ -76,6 +77,14 @@ public class AcceptOnceFileListFilterTests { doTestRollback(filter); } + @Test + public void testRollbackComposite() { + AcceptOnceFileListFilter filter = new AcceptOnceFileListFilter(); + CompositeFileListFilter composite = new CompositeFileListFilter( + Collections.singletonList(filter)); + doTestRollback(composite); + } + protected void doTestRollback(ReversibleFileListFilter filter) { String[] files = new String[] {"foo", "bar", "baz"}; List passed = filter.filterFiles(files); diff --git a/spring-integration-sftp/.gitignore b/spring-integration-sftp/.gitignore index cb9bdcd169..ba984ff96e 100644 --- a/spring-integration-sftp/.gitignore +++ b/spring-integration-sftp/.gitignore @@ -1,3 +1,4 @@ local-test-dir/*.test +local-test-dir/rollback/*.txt remote-target-dir/*foo* remote-target-dir/*test* diff --git a/spring-integration-sftp/local-test-dir/readme.txt b/spring-integration-sftp/local-test-dir/readme.txt index c94b0e60d5..e308f5e4ee 100644 --- a/spring-integration-sftp/local-test-dir/readme.txt +++ b/spring-integration-sftp/local-test-dir/readme.txt @@ -1 +1 @@ -Don't delete. This directory is used by test cases \ No newline at end of file +don't delete this dir used for testing \ No newline at end of file diff --git a/spring-integration-sftp/local-test-dir/rollback/sftpSource1.txt b/spring-integration-sftp/local-test-dir/rollback/sftpSource1.txt deleted file mode 100644 index 90607bb217..0000000000 --- a/spring-integration-sftp/local-test-dir/rollback/sftpSource1.txt +++ /dev/null @@ -1 +0,0 @@ -source1 \ No newline at end of file diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml new file mode 100644 index 0000000000..727895111f --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests-context.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java new file mode 100644 index 0000000000..d3acd57400 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/RollbackLocalFilterTests.java @@ -0,0 +1,76 @@ +/* + * Copyright 2015 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.sftp.inbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.1.7 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RollbackLocalFilterTests { + + @Autowired + private Crash crash; + + @Test + public void testRollback() throws Exception { + assertTrue(this.crash.getLatch().await(10, TimeUnit.SECONDS)); + assertEquals("sftpSource1.txt", this.crash.getFile().getName()); + } + + public static class Crash { + + private final CountDownLatch latch = new CountDownLatch(2); + + private final AtomicBoolean shouldCrash = new AtomicBoolean(); + + private volatile File file; + + public CountDownLatch getLatch() { + return latch; + } + + public File getFile() { + return file; + } + + public void handle(File in) { + latch.countDown(); + if (this.shouldCrash.compareAndSet(false, true)) { + throw new RuntimeException("foo"); + } + this.file = in; + } + } + +} diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index affa81faa0..f09bdaa5d5 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -274,6 +274,52 @@ Only then is the poll operation considered complete, and the poller will wait fo You can alternatively set the 'max-messages-per-poll' value to a positive value indicating the upward limit of Messages to be created from files with each poll. For example, a value of 10 means that on each poll it will attempt to process no more than 10 files. +==== Recovering from Failures + +It is important to understand the architecture of the adapter. +There is a file synchronizer which fetches the files, and a `FileReadingMessageSource` to emit a message for each +synchronized file. +As discussed above, there are two filters involved. +The `filter` attribute (and patterns) refers to the remote (FTP) file list - to avoid fetching files that have already +been fetched. +The `local-filter` is used by the `FileReadingMessageSource` to determine which files are to be sent as messages. + +The synchronizer lists the remote files and consults its filter; the files are then transferred. +If an IO error occurs during file transfer, any files that have already been added to the filter are removed so they +are eligible to be re-fetched on the next poll. +This only applies if the filter implements `ReversibleFileListFilter` (such as the `AcceptOnceFileListFilter`). + +If, after synchronizing the files, an error occurs on the downstream flow processing a file, there is __no__ automatic +rollback of the filter so the failed file will __not__ be reprocessed by default. + +If you wish to reprocess such files after a failure, you can use configuration similar to the following to facilitate +the removal of the failed file from the filter. +This will work for any `ResettableFileListFilter`. + +[source, xml] +---- + + + + + + + + + + + + + +---- + [[ftp-outbound]] === FTP Outbound Channel Adapter diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 96be14f24f..55904238d2 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -359,6 +359,52 @@ If you need a custom filter implementation simply include a reference in your ad ---- +==== Recovering from Failures + +It is important to understand the architecture of the adapter. +There is a file synchronizer which fetches the files, and a `FileReadingMessageSource` to emit a message for each +synchronized file. +As discussed above, there are two filters involved. +The `filter` attribute (and patterns) refers to the remote (SFTP) file list - to avoid fetching files that have already +been fetched. +The `local-filter` is used by the `FileReadingMessageSource` to determine which files are to be sent as messages. + +The synchronizer lists the remote files and consults its filter; the files are then transferred. +If an IO error occurs during file transfer, any files that have already been added to the filter are removed so they +are eligible to be re-fetched on the next poll. +This only applies if the filter implements `ReversibleFileListFilter` (such as the `AcceptOnceFileListFilter`). + +If, after synchronizing the files, an error occurs on the downstream flow processing a file, there is _no_ automatic +rollback of the filter so the failed file will _not_ be reprocessed by default. + +If you wish to reprocess such files after a failure, you can use configuration similar to the following to facilitate +the removal of the failed file from the filter. +This will work for any `ResettableFileListFilter`. + +[source, xml] +---- + + + + + + + + + + + + + +---- + [[sftp-outbound]] === SFTP Outbound Channel Adapter