Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4212: FWMH Predicates and DSL: flushWhenIdle

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4212

Add `firstWrite` to predicate methods.
Add DSL support for `flushWhenIdle`

While these are breaking API changes on the interfaces, I feel this is
likely not going to affect many users and this is a major release after
all.

I have added a note to the 5.0 migration guide.
This commit is contained in:
Gary Russell
2017-01-23 11:07:22 -05:00
committed by Artem Bilan
parent 5f450ed959
commit 8b591482c1
6 changed files with 52 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java
View File

@@ -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<String, FileState> 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<String, FileState> 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());

16
spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java
View File

@@ -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.

4
spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java
View File

@@ -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<InputStream>(new ByteArrayInputStream("bux".getBytes())));
called.set(false);
handler.flushIfNeeded((fileAbsolutePath, lastWrite) -> {
handler.flushIfNeeded((fileAbsolutePath, firstWrite, lastWrite) -> {
called.set(true);
return true;
});

6
spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java
View File

@@ -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;
}),

1
src/reference/asciidoc/file.adoc
View File

@@ -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.

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -61,6 +61,9 @@ See <<file-reading>> for more information.
The tail adapter now supports `idleEventInterval` to emit events when there is no data in the file during that period.
See <<file-tailing>> for more information.
The flush predicates for the `FileWritingMessageHandler` now have an additional parameter.
See <<file-flushing>> 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.