diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index 39b9167ed8..44f952b565 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -838,8 +838,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand /** * When using {@link FileExistsMode#APPEND_NO_FLUSH} you can invoke this method to - * selectively flush open files. For each open file the supplied - * {@link MessageFlushPredicate#shouldFlush(String, long, Message)} + * selectively flush and close open files. For each open file the supplied + * {@link MessageFlushPredicate#shouldFlush(String, long, long, Message)} * method is invoked and if true is returned, the file is flushed. * @param flushPredicate the {@link FlushPredicate}. * @since 4.3 @@ -849,7 +849,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand while (iterator.hasNext()) { Entry entry = iterator.next(); FileState state = entry.getValue(); - if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite)) { + if (flushPredicate.shouldFlush(entry.getKey(), state.firstWrite, state.lastWrite)) { iterator.remove(); state.close(); } @@ -858,8 +858,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand /** * When using {@link FileExistsMode#APPEND_NO_FLUSH} you can invoke this method to - * selectively flush open files. For each open file the supplied - * {@link MessageFlushPredicate#shouldFlush(String, long, Message)} + * selectively flush and close open files. For each open file the supplied + * {@link MessageFlushPredicate#shouldFlush(String, long, long, Message)} * method is invoked and if true is returned, the file is flushed. * @param flushPredicate the {@link MessageFlushPredicate}. * @param filterMessage an optional message passed into the predicate. @@ -870,7 +870,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand while (iterator.hasNext()) { Entry entry = iterator.next(); FileState state = entry.getValue(); - if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite, filterMessage)) { + if (flushPredicate.shouldFlush(entry.getKey(), state.firstWrite, state.lastWrite, filterMessage)) { iterator.remove(); state.close(); } @@ -953,41 +953,49 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } /** - * When using {@link FileExistsMode#APPEND_NO_FLUSH} - * an implementation of this interface is called for each file that has pending data - * to flush when {@link FileWritingMessageHandler#flushIfNeeded(FlushPredicate)} - * is invoked. + * When using {@link FileExistsMode#APPEND_NO_FLUSH}, an implementation of this + * interface is called for each file that has pending data to flush and close when + * {@link FileWritingMessageHandler#flushIfNeeded(FlushPredicate)} is invoked. * @since 4.3 * */ + @FunctionalInterface public interface FlushPredicate { /** + * Return true to cause the file to be flushed and closed. * @param fileAbsolutePath the path to the file. - * @param lastWrite the time of the last write - {@link System#currentTimeMillis()}. - * @return true if the file should be flushed. + * @param firstWrite the time of the first write to a new or previously closed + * file. + * @param lastWrite the time of the last write - + * {@link System#currentTimeMillis()}. + * @return true if the file should be flushed and closed. */ - boolean shouldFlush(String fileAbsolutePath, long lastWrite); + boolean shouldFlush(String fileAbsolutePath, long firstWrite, long lastWrite); } /** * When using {@link FileExistsMode#APPEND_NO_FLUSH} * an implementation of this interface is called for each file that has pending data - * to flush. + * to flush when a trigger message is received. * @see FileWritingMessageHandler#trigger(Message) * @since 4.3 * */ + @FunctionalInterface public interface MessageFlushPredicate { /** + * Return true to cause the file to be flushed and closed. * @param fileAbsolutePath the path to the file. + * @param firstWrite the time of the first write to a new or previously closed + * file. * @param lastWrite the time of the last write - {@link System#currentTimeMillis()}. * @param filterMessage an optional message to be used in the decision process. - * @return true if the file should be flushed. + * @return true if the file should be flushed and closed. */ - boolean shouldFlush(String fileAbsolutePath, long lastWrite, Message filterMessage); + boolean shouldFlush(String fileAbsolutePath, long firstWrite, long lastWrite, Message filterMessage); } @@ -1001,7 +1009,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } @Override - public boolean shouldFlush(String fileAbsolutePath, long lastWrite, Message triggerMessage) { + public boolean shouldFlush(String fileAbsolutePath, long firstWrite, long lastWrite, + Message triggerMessage) { Pattern pattern; if (triggerMessage.getPayload() instanceof String) { pattern = Pattern.compile((String) triggerMessage.getPayload()); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java index bc13124194..35693f2505 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -188,12 +188,26 @@ public class FileWritingMessageHandlerSpec * @param flushInterval the interval. * @return the spec. * @see FileWritingMessageHandler#setBufferSize(int) + * @see #flushWhenIdle(boolean) */ public FileWritingMessageHandlerSpec flushInterval(long flushInterval) { this.target.setFlushInterval(flushInterval); return this; } + /** + * Set the flush when idle flag to false if you wish the interval to apply to when + * the file was opened rather than when the file was last written. + * @param flushWhenIdle false to flush if the interval since the file was opened has elapsed. + * @return the spec. + * @see FileWritingMessageHandler#setFlushWhenIdle(boolean) + * @see #flushInterval(long) + */ + public FileWritingMessageHandlerSpec flushWhenIdle(boolean flushWhenIdle) { + this.target.setFlushWhenIdle(flushWhenIdle); + return this; + } + /** * Specify a {@link TaskScheduler} for flush task when the {@link FileExistsMode#APPEND_NO_FLUSH} is in use. * @param taskScheduler the {@link TaskScheduler} to use. diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java index 1f18f47484..3bbc966d72 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java @@ -480,7 +480,7 @@ public class FileWritingMessageHandlerTests { handler.setFlushInterval(30000); final AtomicBoolean called = new AtomicBoolean(); - handler.setFlushPredicate((fileAbsolutePath, lastWrite, triggerMessage) -> { + handler.setFlushPredicate((fileAbsolutePath, firstWrite, lastWrite, triggerMessage) -> { called.set(true); return true; }); @@ -491,7 +491,7 @@ public class FileWritingMessageHandlerTests { handler.handleMessage(new GenericMessage(new ByteArrayInputStream("bux".getBytes()))); called.set(false); - handler.flushIfNeeded((fileAbsolutePath, lastWrite) -> { + handler.flushIfNeeded((fileAbsolutePath, firstWrite, lastWrite) -> { called.set(true); return true; }); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java index f64b8c8f42..ceab398001 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java @@ -162,6 +162,8 @@ public class FileTests { } } DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("flushWhenIdle")); + assertEquals(60000L, dfa.getPropertyValue("flushInterval")); dfa.setPropertyValue("fileNameGenerator", fileNameGenerator); this.fileFlow1Input.send(message); @@ -331,7 +333,9 @@ public class FileTests { .handle(Files.outboundAdapter(tmpDir.getRoot()) .fileNameGenerator(message -> null) .fileExistsMode(FileExistsMode.APPEND_NO_FLUSH) - .flushPredicate((fileAbsolutePath, lastWrite, filterMessage) -> { + .flushInterval(60000) + .flushWhenIdle(false) + .flushPredicate((fileAbsolutePath, firstWrite, lastWrite, filterMessage) -> { flushPredicateCalled().countDown(); return true; }), diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index ac00ae0b83..52287238ba 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -660,6 +660,7 @@ or `FileWritingMessageHandler.MessageFlushPredicate` implementation. The predicates are called for each open file. See the java docs for these interfaces for more information. +Note that, since _version 5.0_, the predicate methods provide another parameter - the time that the current file was first written to if new or previously closed. When using `flushInterval`, the interval starts at the last write - the file is flushed only if it is idle for the interval. Starting with _version 4.3.7_, and additional property `flushWhenIdle` can be set to `false`, meaning that the interval starts with the first write to a previously flushed (or new) file. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e60b0f1525..81a95d0489 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -61,6 +61,9 @@ See <> for more information. The tail adapter now supports `idleEventInterval` to emit events when there is no data in the file during that period. See <> for more information. +The flush predicates for the `FileWritingMessageHandler` now have an additional parameter. +See <> for more information. + ==== (S)FTP Changes The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.