INT-3721: Add Flush Support to Persistent Filters

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

Support flushing metadata after each update.

Polishing JavaDocs

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.file.filters;
 
 import java.io.Closeable;
+import java.io.Flushable;
 import java.io.IOException;
 import java.util.List;
 
@@ -38,8 +39,12 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 
 	protected final ConcurrentMetadataStore store;
 
+	protected final Flushable flushableStore;
+
 	protected final String prefix;
 
+	protected volatile boolean flushOnUpdate;
+
 	private final Object monitor = new Object();
 
 	public AbstractPersistentAcceptOnceFileListFilter(ConcurrentMetadataStore store, String prefix) {
@@ -47,6 +52,21 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 		Assert.notNull(prefix, "'prefix' cannot be null");
 		this.store = store;
 		this.prefix = prefix;
+		if (store instanceof Flushable) {
+			this.flushableStore = (Flushable) store;
+		}
+		else {
+			this.flushableStore = null;
+		}
+	}
+
+	/**
+	 * Determine whether the metadataStore should be flushed on each update (if {@link Flushable}).
+	 * @param flushOnUpdate true to flush.
+	 * @since 1.4.5
+	 */
+	public void setFlushOnUpdate(boolean flushOnUpdate) {
+		this.flushOnUpdate = flushOnUpdate;
 	}
 
 	@Override
@@ -56,10 +76,17 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 			String newValue = value(file);
 			String oldValue = this.store.putIfAbsent(key, newValue);
 			if (oldValue == null) { // not in store
+				flushIfNeeded();
 				return true;
 			}
 			// same value in store
-			return !isEqual(file, oldValue) && this.store.replace(key, oldValue, newValue);
+			if (!isEqual(file, oldValue)) {
+				if (this.store.replace(key, oldValue, newValue)) {
+					flushIfNeeded();
+					return true;
+				};
+			}
+			return false;
 		}
 	}
 
@@ -76,6 +103,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 			}
 			if (rollingBack) {
 				this.store.remove(buildKey(fileToRollback));
+				flushIfNeeded();
 			}
 		}
 	}
@@ -116,6 +144,22 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 		return this.prefix + this.fileName(file);
 	}
 
+	/**
+	 * Flush the store if it's a {@link Flushable} and
+	 * {@link #setFlushOnUpdate(boolean) flushOnUpdate} is true.
+	 * @since 1.4.5
+	 */
+	protected void flushIfNeeded() {
+		if (this.flushOnUpdate && this.flushableStore != null) {
+			try {
+				this.flushableStore.flush();
+			}
+			catch (IOException e) {
+				// store's responsibility to log
+			}
+		}
+	}
+
 	protected abstract long modified(F file);
 
 	protected abstract String fileName(F file);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2013-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.
@@ -17,9 +17,13 @@
 package org.springframework.integration.file.filters;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
+import java.io.Closeable;
 import java.io.File;
+import java.io.Flushable;
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.Callable;
@@ -28,6 +32,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
@@ -93,6 +98,7 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 		assertEquals(Integer.valueOf(0), theResult); // lost the race, key changed
 
 		file.delete();
+		filter.close();
 	}
 
 	@Override
@@ -115,7 +121,7 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 	}
 
 	@Test
-	public void testRollbackFileSystem() {
+	public void testRollbackFileSystem() throws Exception {
 		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(
 				new SimpleMetadataStore(), "rollback:");
 		File[] files = new File[] {new File("foo"), new File("bar"), new File("baz")};
@@ -130,6 +136,66 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 		assertEquals("baz", now.get(1).getName());
 		now = filter.filterFiles(files);
 		assertEquals(0, now.size());
+		filter.close();
+	}
+
+	@Test
+	/*
+	 * INT-3721: Test all operations that can cause the metadata to be flushed.
+	 */
+	public void testFlush() throws Exception {
+		final AtomicInteger flushes = new AtomicInteger();
+		final AtomicBoolean replaced = new AtomicBoolean();
+		class MS extends SimpleMetadataStore implements Flushable, Closeable {
+
+			@Override
+			public void flush() throws IOException {
+				flushes.incrementAndGet();
+			}
+
+			@Override
+			public void close() throws IOException {
+				flush();
+			}
+
+			@Override
+			public boolean replace(String key, String oldValue, String newValue) {
+				replaced.set(true);
+				return super.replace(key, oldValue, newValue);
+			}
+
+		}
+		MS store = new MS();
+		String prefix = "flush:";
+		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(
+				store, prefix);
+		final File file = File.createTempFile("foo", ".txt");
+		File[] files = new File[] { file };
+		List<File> passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		filter.rollback(passed.get(0), passed);
+		assertEquals(0, flushes.get());
+		filter.setFlushOnUpdate(true);
+		passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		assertEquals(1, flushes.get());
+		filter.rollback(passed.get(0), passed);
+		assertEquals(2, flushes.get());
+		passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		assertEquals(3, flushes.get());
+		passed = filter.filterFiles(files);
+		assertEquals(0, passed.size());
+		assertEquals(3, flushes.get());
+		assertFalse(replaced.get());
+		store.put(prefix + file.getAbsolutePath(), "1");
+		passed = filter.filterFiles(files);
+		assertTrue(Arrays.equals(files, passed.toArray()));
+		assertEquals(4, flushes.get());
+		assertTrue(replaced.get());
+		file.delete();
+		filter.close();
+		assertEquals(5, flushes.get());
 	}
 
 }