INT-4212: FileWritingMessageHandler.flushWhenIdle

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

Add an option to flush after the `flushInterval`, regardless of intermediate writes.

Rename `lastFlush` to `firstWrite`

Polishing

**Cherry-pick to 4.3.x**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -143,6 +143,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 	private volatile long flushInterval = DEFAULT_FLUSH_INTERVAL;
 
+	private volatile boolean flushWhenIdle = true;
+
 	private volatile ScheduledFuture<?> flushTask;
 
 	private volatile MessageFlushPredicate flushPredicate = new DefaultFlushPredicate();
@@ -303,12 +305,27 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 	 * {@code flushInterval} and {@code flushInterval * 1.33} with an average of
 	 * {@code flushInterval * 1.167}.
 	 * @param flushInterval the interval.
+	 * @see #setFlushWhenIdle(boolean)
 	 * @since 4.3
 	 */
 	public void setFlushInterval(long flushInterval) {
 		this.flushInterval = flushInterval;
 	}
 
+	/**
+	 * Determine whether the {@link #setFlushInterval(long) flushInterval} applies only
+	 * to idle files (default) or whether to flush on that interval after the first
+	 * write to a previously flushed or new file.
+	 * @param flushWhenIdle false to flush on the interval after the first write
+	 * to a closed file.
+	 * @see #setFlushInterval(long)
+	 * @see #setBufferSize(int)
+	 * @since 4.3.7
+	 */
+	public void setFlushWhenIdle(boolean flushWhenIdle) {
+		this.flushWhenIdle = flushWhenIdle;
+	}
+
 	@Override
 	public void setTaskScheduler(TaskScheduler taskScheduler) {
 		super.setTaskScheduler(taskScheduler);
@@ -876,6 +893,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 		private final BufferedOutputStream stream;
 
+		private final long firstWrite = System.currentTimeMillis();
+
 		private volatile long lastWrite;
 
 		FileState(BufferedWriter writer) {
@@ -919,7 +938,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 				while (iterator.hasNext()) {
 					Entry<String, FileState> entry = iterator.next();
 					FileState state = entry.getValue();
-					if (state.lastWrite < expired) {
+					if (state.lastWrite < expired ||
+							(!FileWritingMessageHandler.this.flushWhenIdle && state.firstWrite < expired)) {
 						iterator.remove();
 						state.close();
 						if (FileWritingMessageHandler.this.logger.isDebugEnabled()) {

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -73,6 +73,7 @@ abstract class FileWritingMessageHandlerBeanDefinitionBuilder {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "buffer-size");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "flush-interval");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "flush-when-idle");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "flush-predicate");
 		String remoteFileNameGenerator = element.getAttribute("filename-generator");
 		String remoteFileNameGeneratorExpression = element.getAttribute("filename-generator-expression");

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -69,6 +69,8 @@ public class FileWritingMessageHandlerFactoryBean
 
 	private volatile Long flushInterval;
 
+	private volatile Boolean flushWhenIdle;
+
 	private volatile MessageFlushPredicate flushPredicate;
 
 	public void setFileExistsMode(String fileExistsModeAsString) {
@@ -127,6 +129,10 @@ public class FileWritingMessageHandlerFactoryBean
 		this.flushInterval = flushInterval;
 	}
 
+	public void setFlushWhenIdle(boolean flushWhenIdle) {
+		this.flushWhenIdle = flushWhenIdle;
+	}
+
 	public void setFlushPredicate(MessageFlushPredicate flushPredicate) {
 		this.flushPredicate = flushPredicate;
 	}
@@ -183,6 +189,9 @@ public class FileWritingMessageHandlerFactoryBean
 		if (this.flushInterval != null) {
 			handler.setFlushInterval(this.flushInterval);
 		}
+		if (this.flushWhenIdle != null) {
+			handler.setFlushWhenIdle(this.flushWhenIdle);
+		}
 		if (this.flushPredicate != null) {
 			handler.setFlushPredicate(this.flushPredicate);
 		}

spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd

@@ -549,12 +549,26 @@ Only files matching this regular expression will be picked up by this adapter.
 		<xsd:attribute name="flush-interval" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
-					When using 'mode=APPEND_NO_FLUSH' if this time (ms) elapses
+					When using 'mode=APPEND_NO_FLUSH', if this time (ms) elapses
 					without any new writes, the data is flushed and the file closed.
 					Default 30000.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="flush-when-idle" default="true">
+			<xsd:annotation>
+				<xsd:documentation>
+					When using 'mode=APPEND_NO_FLUSH', set to false to indicate the
+					'flush-interval' starts from the first new write to a previously
+					flushed (or new) file. When true, the interval starts from the
+					last write (the file is flushed if it has no writes during the interval).
+					Default true.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attribute name="flush-predicate" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -29,7 +29,11 @@ import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.startsWith;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 import java.io.ByteArrayInputStream;
 import java.io.File;
@@ -39,14 +43,17 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Matcher;
 
+import org.apache.commons.logging.Log;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -490,6 +497,25 @@ public class FileWritingMessageHandlerTests {
 		});
 		assertThat(file.length(), equalTo(24L));
 		assertTrue(called.get());
+
+		handler.stop();
+		Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class));
+		new DirectFieldAccessor(handler).setPropertyValue("logger", logger);
+		when(logger.isDebugEnabled()).thenReturn(true);
+		final AtomicInteger flushes = new AtomicInteger();
+		doAnswer(i -> {
+			flushes.incrementAndGet();
+			return null;
+		}).when(logger).debug(startsWith("Flushed:"));
+		handler.setFlushInterval(50);
+		handler.setFlushWhenIdle(false);
+		handler.start();
+		for (int i = 0; i < 40; i++) {
+			handler.handleMessage(new GenericMessage<String>("foo"));
+			Thread.sleep(5);
+		}
+		assertThat(flushes.get(), greaterThanOrEqualTo(2));
+		handler.stop();
 	}
 
 	void assertFileContentIsMatching(Message<?> result) throws IOException {

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml

@@ -21,6 +21,7 @@
 
 	<file:outbound-channel-adapter id="adapterWithCustomNameGenerator"
 								   channel="testChannel"
+								   flush-when-idle="false"
 								   filename-generator="customFileNameGenerator"
 								   directory="${java.io.tmpdir}"/>
 

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -139,6 +139,7 @@ public class FileOutboundChannelAdapterParserTests {
 		assertNotNull(expression);
 		assertEquals("'foo.txt'", expression.getExpressionString());
 		assertEquals(Boolean.FALSE, handlerAccessor.getPropertyValue("deleteSourceFiles"));
+		assertEquals(Boolean.TRUE, handlerAccessor.getPropertyValue("flushWhenIdle"));
 	}
 
 	@Test
@@ -156,6 +157,7 @@ public class FileOutboundChannelAdapterParserTests {
 		assertEquals(expected, actual);
 		assertTrue(handlerAccessor.getPropertyValue("fileNameGenerator") instanceof CustomFileNameGenerator);
 		assertEquals(".writing", handlerAccessor.getPropertyValue("temporaryFileSuffix"));
+		assertEquals(Boolean.FALSE, handlerAccessor.getPropertyValue("flushWhenIdle"));
 	}
 
 	@Test

src/reference/asciidoc/file.adoc

@@ -661,6 +661,9 @@ or `FileWritingMessageHandler.MessageFlushPredicate` implementation.
 The predicates are called for each open file.
 See the java docs for these interfaces for more information.
 
+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.
+
 [[file-timestamps]]
 ==== File Timestamps