INT-3464 Exceptions and AcceptOnceFileListFilter

JIRA: https://jira.spring.io/browse/INT-3464

Previously, if an IO exeption occurred while synchronizing
files, and an AcceptOnceFileListFilter is being used, files that were
not transferred would not be fetched next time.

Add strategy `ReversibleFileListFilter` that can rollback previously
accepted files.

Implement this interface on `AcceptOnceFileListFilter` and
`AbstractPersistentAcceptOnceFileListFilter`.

Add test cases.

Polishing

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java

@@ -13,8 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.file.filters;
 
+import java.util.List;
+
 import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.util.Assert;
 
@@ -28,7 +31,8 @@ import org.springframework.util.Assert;
  * @since 3.0
  *
  */
-public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> {
+public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F>
+		implements ReversibleFileListFilter<F> {
 
 	protected final ConcurrentMetadataStore store;
 
@@ -52,10 +56,25 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 			if (oldValue == null) { // not in store
 				return true;
 			}
-			if (isEqual(file, oldValue)) { // same value in store
-				return false;
+			// same value in store
+			return !isEqual(file, oldValue) && this.store.replace(key, oldValue, newValue);
+		}
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * @since 4.0.4
+	 */
+	@Override
+	public void rollback(F file, List<F> files) {
+		boolean rollingBack = false;
+		for (F fileToRollback : files) {
+			if (fileToRollback.equals(file)) {
+				rollingBack = true;
+			}
+			if (rollingBack) {
+				this.store.remove(buildKey(fileToRollback));
 			}
-			return this.store.replace(key, oldValue, newValue); // true if replace successful
 		}
 	}
 

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.file.filters;
 
+import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.LinkedBlockingQueue;
 
@@ -28,9 +29,10 @@ import java.util.concurrent.LinkedBlockingQueue;
  *
  * @author Iwein Fuld
  * @author Josh Long
+ * @author Gary Russell
  * @since 1.0.0
  */
-public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> {
+public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> implements ReversibleFileListFilter<F> {
 
 	private final Queue<F> seen;
 
@@ -56,6 +58,7 @@ public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> {
 	}
 
 
+	@Override
 	public boolean accept(F file) {
 		synchronized (this.monitor) {
 			if (this.seen.contains(file)) {
@@ -69,4 +72,21 @@ public class AcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> {
 		}
 	}
 
+	/**
+	 * {@inheritDoc}
+	 * @since 4.0.4
+	 */
+	@Override
+	public void rollback(F file, List<F> files) {
+		boolean rollingBack = false;
+		for (F fileToRollback : files) {
+			if (fileToRollback.equals(file)) {
+				rollingBack = true;
+			}
+			if (rollingBack) {
+				this.seen.remove(fileToRollback);
+			}
+		}
+	}
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/filters/ReversibleFileListFilter.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2014 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;
+
+import java.util.List;
+
+/**
+ *
+ * A {@link FileListFilter} that allows the caller to reverse (roll back) state
+ * changes.
+ *
+ * @author Gary Russell
+ * @since 4.0.4
+ *
+ */
+public interface ReversibleFileListFilter<F> extends FileListFilter<F> {
+
+	/**
+	 * Indicate that not all files previously passed by this filter (in {@link #filterFiles(Object[])}
+	 * have been processed; the file must be in the list of files; it, and all files after it, will
+	 * be considered to have not been processed and will be considered next time.
+	 * @param file the file which failed.
+	 * @param files the list of files that were returned by {@link #filterFiles(Object[])}.
+	 */
+	void rollback(F file, List<F> files);
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java

@@ -22,7 +22,6 @@ import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.List;
 
 import org.apache.commons.logging.Log;
@@ -33,6 +32,7 @@ import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.expression.IntegrationEvaluationContextAware;
 import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.ReversibleFileListFilter;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.session.Session;
@@ -170,12 +170,28 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
 				public Integer doInSession(Session<F> session) throws IOException {
 					F[] files = session.list(AbstractInboundFileSynchronizer.this.remoteDirectory);
 					if (!ObjectUtils.isEmpty(files)) {
-						Collection<F> filteredFiles = AbstractInboundFileSynchronizer.this.filterFiles(files);
+						List<F> filteredFiles = AbstractInboundFileSynchronizer.this.filterFiles(files);
 						for (F file : filteredFiles) {
-							if (file != null) {
-								AbstractInboundFileSynchronizer.this.copyFileToLocalDirectory(
-										AbstractInboundFileSynchronizer.this.remoteDirectory, file, localDirectory,
-										session);
+							try {
+								if (file != null) {
+									AbstractInboundFileSynchronizer.this.copyFileToLocalDirectory(
+											AbstractInboundFileSynchronizer.this.remoteDirectory, file, localDirectory,
+											session);
+								}
+							}
+							catch (RuntimeException e) {
+								if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
+									((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
+											.rollback(file, filteredFiles);
+								}
+								throw e;
+							}
+							catch (IOException e) {
+								if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
+									((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
+											.rollback(file, filteredFiles);
+								}
+								throw e;
 							}
 						}
 						return filteredFiles.size();
@@ -194,7 +210,8 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
 		}
 	}
 
-	private void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory, Session<F> session) throws IOException {
+	protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
+			Session<F> session) throws IOException {
 		String remoteFileName = this.getFilename(remoteFile);
 		String localFileName = this.generateLocalFileName(remoteFileName);
 		String remoteFilePath = remoteDirectoryPath + remoteFileSeparator + remoteFileName;

spring-integration-file/src/test/java/org/springframework/integration/file/filters/AcceptOnceFileListFilterTests.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2014 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+
+/**
+ * @author Gary Russell
+ * @since 4.0.4
+ *
+ */
+public class AcceptOnceFileListFilterTests {
+
+	@Test
+	public void testRollback() {
+		AcceptOnceFileListFilter<String> filter = new AcceptOnceFileListFilter<String>();
+		doTestRollback(filter);
+	}
+
+	protected void doTestRollback(ReversibleFileListFilter<String> filter) {
+		String[] files = new String[] {"foo", "bar", "baz"};
+		List<String> passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		List<String> now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+		filter.rollback(passed.get(1), passed);
+		now = filter.filterFiles(files);
+		assertEquals(2, now.size());
+		assertEquals("bar", now.get(0));
+		assertEquals("baz", now.get(1));
+		now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java

@@ -13,12 +13,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.file.filters;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
@@ -36,7 +39,7 @@ import org.springframework.integration.metadata.SimpleMetadataStore;
  * @since 3.0
  *
  */
-public class PersistentAcceptOnceFileListFilterTests {
+public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListFilterTests {
 
 	@Test
 	public void testFileSystem() throws Exception {
@@ -92,4 +95,41 @@ public class PersistentAcceptOnceFileListFilterTests {
 		file.delete();
 	}
 
+	@Override
+	@Test
+	public void testRollback() {
+		AbstractPersistentAcceptOnceFileListFilter<String> filter = new AbstractPersistentAcceptOnceFileListFilter<String>(
+				new SimpleMetadataStore(), "rollback:") {
+
+					@Override
+					protected long modified(String file) {
+						return 0;
+					}
+
+					@Override
+					protected String fileName(String file) {
+						return file;
+					}
+				};
+		doTestRollback(filter);
+	}
+
+	@Test
+	public void testRollbackFileSystem() {
+		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(
+				new SimpleMetadataStore(), "rollback:");
+		File[] files = new File[] {new File("foo"), new File("bar"), new File("baz")};
+		List<File> passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		List<File> now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+		filter.rollback(passed.get(1), passed);
+		now = filter.filterFiles(files);
+		assertEquals(2, now.size());
+		assertEquals("bar", now.get(0).getName());
+		assertEquals("baz", now.get(1).getName());
+		now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+	}
+
 }

spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2014 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.remote.synchronizer;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
+import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.messaging.MessagingException;
+
+/**
+ * @author Gary Russell
+ * @since 4.0.4
+ *
+ */
+public class AbstractRemoteFileSynchronizerTests {
+
+	@Test
+	public void testRollback() {
+		final AtomicBoolean failWhenCopyingBar = new AtomicBoolean(true);
+		final AtomicInteger count = new AtomicInteger();
+		SessionFactory<String> sf = new StringSessionFactory();
+		AbstractInboundFileSynchronizer<String> sync = new AbstractInboundFileSynchronizer<String>(sf) {
+
+			@Override
+			protected boolean isFile(String file) {
+				return true;
+			}
+
+			@Override
+			protected String getFilename(String file) {
+				return file;
+			}
+
+			@Override
+			protected long getModified(String file) {
+				return 0;
+			}
+
+			@Override
+			protected void copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, File localDirectory,
+					Session<String> session) throws IOException {
+				if ("bar".equals(remoteFile) && failWhenCopyingBar.getAndSet(false)) {
+					throw new IOException("fail");
+				}
+				count.incrementAndGet();
+			}
+
+		};
+		sync.setFilter(new AcceptOnceFileListFilter<String>());
+
+		try {
+			sync.synchronizeToLocalDirectory(mock(File.class));
+			assertEquals(1, count.get());
+			fail("Expected exception");
+		}
+		catch (MessagingException e) {
+			assertThat(e.getCause(), instanceOf(MessagingException.class));
+			assertThat(e.getCause().getCause(), instanceOf(IOException.class));
+			assertEquals("fail", e.getCause().getCause().getMessage());
+		}
+		sync.synchronizeToLocalDirectory(mock(File.class));
+		assertEquals(3, count.get());
+	}
+
+	private class StringSessionFactory implements SessionFactory<String> {
+
+		@Override
+		public Session<String> getSession() {
+			return new StringSession();
+		}
+
+	}
+
+	private class StringSession implements Session<String> {
+
+		@Override
+		public boolean remove(String path) throws IOException {
+			return true;
+		}
+
+		@Override
+		public String[] list(String path) throws IOException {
+			return new String[] {"foo", "bar", "baz"};
+		}
+
+		@Override
+		public void read(String source, OutputStream outputStream) throws IOException {
+		}
+
+		@Override
+		public void write(InputStream inputStream, String destination) throws IOException {
+		}
+
+		@Override
+		public boolean mkdir(String directory) throws IOException {
+			return true;
+		}
+
+		@Override
+		public void rename(String pathFrom, String pathTo) throws IOException {
+		}
+
+		@Override
+		public void close() {
+		}
+
+		@Override
+		public boolean isOpen() {
+			return true;
+		}
+
+		@Override
+		public boolean exists(String path) throws IOException {
+			return true;
+		}
+
+		@Override
+		public String[] listNames(String path) throws IOException {
+			return new String[0];
+		}
+
+		@Override
+		public InputStream readRaw(String source) throws IOException {
+			return null;
+		}
+
+		@Override
+		public boolean finalizeRaw() throws IOException {
+			return true;
+		}
+
+	}
+
+}

spring-integration-ftp/.gitignore

@@ -1,3 +1,2 @@
 local-test-dir/*.test
 remote-target-dir/*test
-test

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/filters/FtpPersistentAcceptOnceFileListFiltertests.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2014 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.ftp.filters;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.List;
+
+import org.apache.commons.net.ftp.FTPFile;
+import org.junit.Test;
+
+import org.springframework.integration.metadata.SimpleMetadataStore;
+
+/**
+ * @author Gary Russell
+ * @since 4.0.4
+ *
+ */
+public class FtpPersistentAcceptOnceFileListFilterTests {
+
+	@Test
+	public void testRollback() {
+		FtpPersistentAcceptOnceFileListFilter filter = new FtpPersistentAcceptOnceFileListFilter(
+				new SimpleMetadataStore(), "rollback:");
+		FTPFile ftpFile1 = new FTPFile();
+		ftpFile1.setName("foo");
+		ftpFile1.setTimestamp(Calendar.getInstance());
+		FTPFile ftpFile2 = new FTPFile();
+		ftpFile2.setName("bar");
+		ftpFile2.setTimestamp(Calendar.getInstance());
+		FTPFile ftpFile3 = new FTPFile();
+		ftpFile3.setName("baz");
+		ftpFile3.setTimestamp(Calendar.getInstance());
+		FTPFile[] files = new FTPFile[] {ftpFile1, ftpFile2, ftpFile3};
+		List<FTPFile> passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		List<FTPFile> now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+		filter.rollback(passed.get(1), passed);
+		now = filter.filterFiles(files);
+		assertEquals(2, now.size());
+		assertEquals("bar", now.get(0).getName());
+		assertEquals("baz", now.get(1).getName());
+		now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+	}
+
+}

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFiltertests.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2014 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.filters;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.List;
+
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+import com.jcraft.jsch.SftpATTRS;
+import org.junit.Test;
+
+import org.springframework.integration.metadata.SimpleMetadataStore;
+
+/**
+ * @author Gary Russell
+ * @since 4.0.4
+ *
+ */
+public class SftpPersistentAcceptOnceFileListFilterTests {
+
+	@Test
+	public void testRollback() throws Exception {
+		SftpPersistentAcceptOnceFileListFilter filter = new SftpPersistentAcceptOnceFileListFilter(
+				new SimpleMetadataStore(), "rollback:");
+		ChannelSftp channel = new ChannelSftp();
+		SftpATTRS attrs = mock(SftpATTRS.class);
+		@SuppressWarnings("unchecked")
+		Constructor<LsEntry> ctor = (Constructor<LsEntry>) LsEntry.class.getDeclaredConstructors()[0];
+		ctor.setAccessible(true);
+		LsEntry sftpFile1 = ctor.newInstance(channel, "foo", "foo", attrs);
+		LsEntry sftpFile2 = ctor.newInstance(channel, "bar", "bar", attrs);
+		LsEntry ftpFile3 = ctor.newInstance(channel, "baz", "baz", attrs);
+		LsEntry[] files = new LsEntry[] {sftpFile1, sftpFile2, ftpFile3};
+		List<LsEntry> passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		List<LsEntry> now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+		filter.rollback(passed.get(1), passed);
+		now = filter.filterFiles(files);
+		assertEquals(2, now.size());
+		assertEquals("bar", now.get(0).getFilename());
+		assertEquals("baz", now.get(1).getFilename());
+		now = filter.filterFiles(files);
+		assertEquals(0, now.size());
+	}
+
+}