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());
 	}
 
 }

src/reference/asciidoc/file.adoc

@@ -37,6 +37,9 @@ This filter matches on the filename and modified time.
 
 Since _version 4.0_, this filter requires a `ConcurrentMetadataStore`.
 When used with a shared data store (such as `Redis` with the `RedisMetadataStore`) this allows filter keys to be shared across multiple application instances, or when a network file share is being used by multiple servers.
+
+Since __version 4.1.5__, this filter has a new property `flushOnUpdate` which will cause it to flush the
+metadata store on every update (if the store implements `Flushable`).
 =====
 
 [source,xml]

src/reference/asciidoc/ftp.adoc

@@ -180,6 +180,9 @@ Use the `local-filter` attribute to configure the behavior of the local file sys
 To solve these particular use cases, you can use a`FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead.
 This filter also stores the accepted file names and modified timestamp in an instance of the`MetadataStore` strategy (<<metadata-store>>), and will detect the change in the local file modified time.
 
+Since __version 4.1.5__, these filters have a new property `flushOnUpdate` which will cause them to flush the
+metadata store on every update (if the store implements `Flushable`).
+
 IMPORTANT: Further, if you use a distributed `MetadataStore` (such as <<redis-metadata-store>> or <<gemfire-metadata-store>>) you can have multiple instances of the same adapter/application and be sure that one and only one will process a file.
 
 The actual local filter is a `CompositeFileListFilter` containing the supplied filter and a pattern filter that prevents processing files that are in the process of being downloaded (based on the `temporary-file-suffix`); files are downloaded with this suffix (default: `.writing`) and the file is renamed to its final name when the transfer is complete, making it 'visible' to the filter.

src/reference/asciidoc/meta-data-store.adoc

@@ -22,6 +22,9 @@ the framework:
 
 The `PropertiesPersistingMetadataStore` is backed by a properties file and a http://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/util/PropertiesPersister.html[PropertiesPersister].
 
+By default, it only persists the state when the application context is closed normally. It implements `Flushable` so you
+can persist the state at will, be invoking `flush()`.
+
 [source,xml]
 ----
 <bean id="metadataStore"

src/reference/asciidoc/sftp.adoc

@@ -266,6 +266,9 @@ Use the `local-filter` attribute to configure the behavior of the local file sys
 To solve these particular use cases, you can use a`FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead.
 This filter also stores the accepted file names and modified timestamp in an instance of the`MetadataStore` strategy (<<metadata-store>>), and will detect the change in the local file modified time.
 
+Since __version 4.1.5__, these filters have a new property `flushOnUpdate` which will cause them to flush the
+metadata store on every update (if the store implements `Flushable`).
+
 IMPORTANT: Further, if you use a distributed `MetadataStore` (such as <<redis-metadata-store>> or <<gemfire-metadata-store>>) you can have multiple instances of the same adapter/application and be sure that one and only one will process a file.
 
 The actual local filter is a `CompositeFileListFilter` containing the supplied filter and a pattern filter that prevents processing files that are in the process of being downloaded (based on the `temporary-file-suffix`); files are downloaded with this suffix (default: `.writing`) and the file is renamed to its final name when the transfer is complete, making it 'visible' to the filter.

src/reference/asciidoc/whats-new.adoc

@@ -109,7 +109,7 @@ attributes with similar purpose as for `<int-amqp:outbound-channel-adapter>`.
 See <<amqp>> for more information.
 
 [[x4.2-xpath-splitter]]
-==== XPath Splitter improvements
+==== XPath Splitter Improvements
 
 The `XPathMessageSplitter` (`<int-xml:xpath-splitter>`) now allows the configuration of `output-properties`
 for the internal `javax.xml.transform.Transformer` and supports an `Iterator` mode (defaults to `true`) for the xpath
@@ -118,9 +118,15 @@ evaluation `org.w3c.dom.NodeList` result.
 See <<xml-xpath-splitting>> for more information.
 
 [[x4.2-http-changes]]
-==== HTTP changes
+==== HTTP Changes
 
 The HTTP Inbound Endpoints (`<int-http:inbound-channel-adapter>` and `<int-http:inbound-gateway>`) now allow the
 configuration of _Cross-Origin Resource Sharing (CORS)_.
 
 See <<cors>> for more information.
+
+[[x4.2-file-filter]]
+==== Persistent File List Filter Changes
+
+The `AbstractPersistentFileListFilter` has a new property `flushOnUpdate` which, when set to true, will `flush()` the
+metadata store if it implements `Flushable` (e.g. the `PropertiesPersistenMetadataStore`).