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 fb3e688f21..2627792215 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/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; + } + } + +}