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 d07354ac12..fb3e688f21 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 @@ -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 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 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 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 extends Abst } if (rollingBack) { this.store.remove(buildKey(fileToRollback)); + flushIfNeeded(); } } } @@ -116,6 +144,22 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter 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); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java index 426a1cd112..ccc1f18a2c 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java +++ b/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 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()); } } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 30b2bbe70c..7604a8db46 100644 --- a/src/reference/asciidoc/file.adoc +++ b/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] diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 500c04ec0e..82b9090247 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/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 (<>), 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 <> or <>) 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. diff --git a/src/reference/asciidoc/meta-data-store.adoc b/src/reference/asciidoc/meta-data-store.adoc index 2fe96bcf59..cfbfa22ae1 100644 --- a/src/reference/asciidoc/meta-data-store.adoc +++ b/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] ---- >), 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 <> or <>) 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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e1893daa80..440fc0d107 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -109,7 +109,7 @@ attributes with similar purpose as for ``. See <> for more information. [[x4.2-xpath-splitter]] -==== XPath Splitter improvements +==== XPath Splitter Improvements The `XPathMessageSplitter` (``) 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 <> for more information. [[x4.2-http-changes]] -==== HTTP changes +==== HTTP Changes The HTTP Inbound Endpoints (`` and ``) now allow the configuration of _Cross-Origin Resource Sharing (CORS)_. See <> 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`).