INT-3574: File Outbound - Don't Flush File

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

Initial commit.

Polishing

Prevent the flusher task from closing while a write is in process.

Add MessageTriggerAction

Flush file(s) on demand.

Polishing - PR Comments

Add FlushPredicate

Polishing - PR Comments

Polishing - More PR Comments and Schema

Polish and Namespace

Docs

flushIfNeeded Must be Synchronized

Fix Race on Stop

Flush immediately after the write if the handler has been stopped.

Remove states from the internal store if handler is stopped.

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

@@ -21,23 +21,32 @@ import java.io.BufferedOutputStream;
 import java.io.BufferedWriter;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStreamWriter;
 import java.nio.charset.Charset;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ScheduledFuture;
 import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.context.Lifecycle;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.MessageTriggerAction;
 import org.springframework.integration.support.locks.DefaultLockRegistry;
 import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.integration.support.locks.PassThruLockRegistry;
@@ -45,6 +54,7 @@ import org.springframework.integration.util.WhileLockedProcessor;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 import org.springframework.util.StreamUtils;
 import org.springframework.util.StringUtils;
@@ -69,6 +79,15 @@ import org.springframework.util.StringUtils;
  * Likewise, any Object can be converted to a String based on its
  * <code>toString()</code> method by the
  * {@link org.springframework.integration.transformer.ObjectToStringTransformer}.
+ * <p>
+ * {@link FileExistsMode#APPEND} adds content to an existing file; the file is closed after
+ * each write.
+ * {@link FileExistsMode#APPEND_NO_FLUSH} adds content to an existing file and the file
+ * is left open without flushing any data. Data will be flushed based on the
+ * {@link #setFlushInterval(long) flushInterval} or when a message is sent to the
+ * {@link #trigger(Message)} method, or a
+ * {@link #flushIfNeeded(MessageFlushPredicate, Message) flushIfNeeded}
+ * method is called.
  *
  * @author Mark Fisher
  * @author Iwein Fuld
@@ -79,10 +98,17 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Tony Falabella
  */
-public class FileWritingMessageHandler extends AbstractReplyProducingMessageHandler {
+public class FileWritingMessageHandler extends AbstractReplyProducingMessageHandler
+		implements Lifecycle, MessageTriggerAction {
 
 	private static final String LINE_SEPARATOR = System.getProperty("line.separator");
 
+	private static final int DEFAULT_BUFFER_SIZE = 8192;
+
+	private static final long DEFAULT_FLUSH_INTERVAL = 30000L;
+
+	private final Map<String, FileState> fileStates = new HashMap<String, FileState>();
+
 	private volatile String temporaryFileSuffix = ".writing";
 
 	private volatile boolean temporaryFileSuffixSet = false;
@@ -111,6 +137,14 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 	private volatile LockRegistry lockRegistry = new PassThruLockRegistry();
 
+	private volatile int bufferSize = DEFAULT_BUFFER_SIZE;
+
+	private volatile long flushInterval = DEFAULT_FLUSH_INTERVAL;
+
+	private volatile ScheduledFuture<?> flushTask;
+
+	private volatile MessageFlushPredicate flushPredicate = new DefaultFlushPredicate();
+
 	/**
 	 * Constructor which sets the {@link #destinationDirectoryExpression} using
 	 * a {@link LiteralExpression}.
@@ -181,7 +215,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 		Assert.notNull(fileExistsMode, "'fileExistsMode' must not be null.");
 		this.fileExistsMode = fileExistsMode;
 
-		if (FileExistsMode.APPEND.equals(fileExistsMode)) {
+		if (FileExistsMode.APPEND.equals(fileExistsMode)
+				|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode)) {
 			this.lockRegistry = this.lockRegistry instanceof PassThruLockRegistry
 					? new DefaultLockRegistry()
 					: this.lockRegistry;
@@ -249,6 +284,42 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 		this.charset = Charset.forName(charset);
 	}
 
+	/**
+	 * Set the buffer size to use while writing to files; default 8192.
+	 * @param bufferSize the buffer size.
+	 * @since 4.3
+	 */
+	public void setBufferSize(int bufferSize) {
+		this.bufferSize = bufferSize;
+	}
+
+	/**
+	 * Set the frequency to flush buffers when {@link FileExistsMode#APPEND_NO_FLUSH} is
+	 * being used.
+	 * @param flushInterval the interval.
+	 * @since 4.3
+	 */
+	public void setFlushInterval(long flushInterval) {
+		this.flushInterval = flushInterval;
+	}
+
+	@Override
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		super.setTaskScheduler(taskScheduler);
+	}
+
+	/**
+	 * Set a {@link MessageFlushPredicate} to use when flushing files when
+	 * {@link FileExistsMode#APPEND_NO_FLUSH} is being used.
+	 * See {@link #trigger(Message)}.
+	 * @param flushPredicate the predicate.
+	 * @since 4.3
+	 */
+	public void setFlushPredicate(MessageFlushPredicate flushPredicate) {
+		Assert.notNull(flushPredicate, "'flushPredicate' cannot be null");
+		this.flushPredicate = flushPredicate;
+	}
+
 	@Override
 	protected void doInit() {
 		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
@@ -259,12 +330,38 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 			validateDestinationDirectory(directory, this.autoCreateDirectory);
 		}
 
-		Assert.state(!(this.temporaryFileSuffixSet && FileExistsMode.APPEND.equals(this.fileExistsMode)),
+		Assert.state(!(this.temporaryFileSuffixSet
+				&& (FileExistsMode.APPEND.equals(this.fileExistsMode)
+						|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode))),
 				"'temporaryFileSuffix' can not be set when appending to an existing file");
 
 		if (!this.fileNameGeneratorSet && this.fileNameGenerator instanceof BeanFactoryAware) {
 			((BeanFactoryAware) this.fileNameGenerator).setBeanFactory(getBeanFactory());
 		}
+
+	}
+
+	@Override
+	public void start() {
+		if (FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode)) {
+			TaskScheduler taskScheduler = getTaskScheduler();
+			Assert.state(taskScheduler != null, "'taskScheduler' is required for FileExistsMode.APPEND_NO_FLUSH");
+			this.flushTask = taskScheduler.scheduleAtFixedRate(new Flusher(), this.flushInterval / 3);
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.flushTask != null) {
+			this.flushTask.cancel(true);
+			this.flushTask = null;
+		}
+		new Flusher().run();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.flushTask != null;
 	}
 
 	private void validateDestinationDirectory(File destinationDirectory, boolean autoCreateDirectory) {
@@ -380,15 +477,20 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 	private File handleInputStreamMessage(final InputStream sourceFileInputStream, File originalFile, File tempFile,
 										  final File resultFile) throws IOException {
-		if (FileExistsMode.APPEND.equals(this.fileExistsMode)) {
-			File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
-			final BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(fileToWriteTo, true));
+		final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode)
+				|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
+
+		if (append) {
+			final File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
+
+			final FileState state = getFileState(fileToWriteTo, false);
 
 			WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry,
 					fileToWriteTo.getAbsolutePath()) {
 
 				@Override
 				protected void whileLocked() throws IOException {
+					BufferedOutputStream bos = state != null ? state.stream : createOutputStream(fileToWriteTo, true);
 					try {
 						byte[] buffer = new byte[StreamUtils.BUFFER_SIZE];
 						int bytesRead = -1;
@@ -398,7 +500,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 						if (FileWritingMessageHandler.this.appendNewLine) {
 							bos.write(LINE_SEPARATOR.getBytes());
 						}
-						bos.flush();
 					}
 					finally {
 						try {
@@ -407,7 +508,15 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 						catch (IOException ex) {
 						}
 						try {
-							bos.close();
+							if (state == null || FileWritingMessageHandler.this.flushTask == null) {
+								bos.close();
+								if (state != null) {
+									fileStates.remove(fileToWriteTo.getAbsolutePath());
+								}
+							}
+							else {
+								state.lastWrite = System.currentTimeMillis();
+							}
 						}
 						catch (IOException ex) {
 						}
@@ -421,7 +530,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 		}
 		else {
 
-			BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(tempFile));
+			BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(tempFile), this.bufferSize);
 
 			try {
 				byte[] buffer = new byte[StreamUtils.BUFFER_SIZE];
@@ -453,16 +562,18 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 	private File handleByteArrayMessage(final byte[] bytes, File originalFile, File tempFile, final File resultFile)
 			throws IOException {
-		File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
+		final File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
+
+		final FileState state = getFileState(fileToWriteTo, false);
 
 		final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode);
 
-		final BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(fileToWriteTo, append));
 		WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry,
 				fileToWriteTo.getAbsolutePath()) {
 
 			@Override
 			protected void whileLocked() throws IOException {
+				BufferedOutputStream bos = state != null ? state.stream : createOutputStream(fileToWriteTo, append);
 				try {
 					bos.write(bytes);
 					if (FileWritingMessageHandler.this.appendNewLine) {
@@ -471,7 +582,15 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 				}
 				finally {
 					try {
-						bos.close();
+						if (state == null || FileWritingMessageHandler.this.flushTask == null) {
+							bos.close();
+							if (state != null) {
+								fileStates.remove(fileToWriteTo.getAbsolutePath());
+							}
+						}
+						else {
+							state.lastWrite = System.currentTimeMillis();
+						}
 					}
 					catch (IOException ex) {
 					}
@@ -486,17 +605,18 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 	private File handleStringMessage(final String content, File originalFile, File tempFile, final File resultFile)
 			throws IOException {
-		File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
+		final File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
+
+		final FileState state = getFileState(fileToWriteTo, true);
 
 		final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode);
 
-		final BufferedWriter writer =
-				new BufferedWriter(new OutputStreamWriter(new FileOutputStream(fileToWriteTo, append), this.charset));
 		WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry,
 				fileToWriteTo.getAbsolutePath()) {
 
 			@Override
 			protected void whileLocked() throws IOException {
+				BufferedWriter writer = state != null ? state.writer : createWriter(fileToWriteTo, append);
 				try {
 					writer.write(content);
 					if (FileWritingMessageHandler.this.appendNewLine) {
@@ -505,7 +625,15 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 				}
 				finally {
 					try {
-						writer.close();
+						if (state == null || FileWritingMessageHandler.this.flushTask == null) {
+							writer.close();
+							if (state != null) {
+								fileStates.remove(fileToWriteTo.getAbsolutePath());
+							}
+						}
+						else {
+							state.lastWrite = System.currentTimeMillis();
+						}
 					}
 					catch (IOException ex) {
 					}
@@ -526,6 +654,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 
 		switch (this.fileExistsMode) {
 			case APPEND:
+			case APPEND_NO_FLUSH:
 				fileToWriteTo = resultFile;
 				break;
 			case FAIL:
@@ -540,7 +669,9 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 	}
 
 	private void cleanUpAfterCopy(File fileToWriteTo, File resultFile, File originalFile) throws IOException {
-		if (!FileExistsMode.APPEND.equals(this.fileExistsMode) && StringUtils.hasText(this.temporaryFileSuffix)) {
+		if (!FileExistsMode.APPEND.equals(this.fileExistsMode)
+				&& !FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode)
+				&& StringUtils.hasText(this.temporaryFileSuffix)) {
 			this.renameTo(fileToWriteTo, resultFile);
 		}
 
@@ -608,4 +739,216 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 		return destinationDirectory;
 	}
 
+	private synchronized FileState getFileState(final File fileToWriteTo, boolean isString)
+			throws FileNotFoundException {
+		String absolutePath = fileToWriteTo.getAbsolutePath();
+		FileState state;
+		boolean appendNoFlush = FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
+		if (appendNoFlush) {
+			state = this.fileStates.get(absolutePath);
+			if (state != null && ((isString && state.stream != null) || (!isString && state.writer != null))) {
+				state.close();
+				state = null;
+				this.fileStates.remove(absolutePath);
+			}
+			if (state == null) {
+				if (isString) {
+					state = new FileState(createWriter(fileToWriteTo, true));
+				}
+				else {
+					state = new FileState(createOutputStream(fileToWriteTo, true));
+				}
+				this.fileStates.put(absolutePath, state);
+			}
+			state.lastWrite = Long.MAX_VALUE; // prevent flush while we write
+		}
+		else {
+			state = null;
+		}
+		return state;
+	}
+
+	private BufferedWriter createWriter(final File fileToWriteTo, final boolean append) throws FileNotFoundException {
+		return new BufferedWriter(new OutputStreamWriter(new FileOutputStream(fileToWriteTo, append), this.charset),
+				this.bufferSize);
+	}
+
+	private BufferedOutputStream createOutputStream(File fileToWriteTo, final boolean append)
+			throws FileNotFoundException {
+		return new BufferedOutputStream(new FileOutputStream(fileToWriteTo, append), this.bufferSize);
+	}
+
+	/**
+	 * When using {@link FileExistsMode#APPEND_NO_FLUSH}, you can send a message to this
+	 * method to flush any file(s) that needs it. By default, the payload must be a regular
+	 * expression ({@link String} or {@link Pattern}) that matches the absolutePath
+	 * of any in-process files. However, if a custom {@link MessageFlushPredicate} is provided,
+	 * the payload can be of any type supported by that implementation.
+	 * @since 4.3
+	 */
+	@Override
+	public void trigger(Message<?> message) {
+		flushIfNeeded(this.flushPredicate, message);
+	}
+
+	/**
+	 * 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)}
+	 * method is invoked and if true is returned, the file is flushed.
+	 * @param flushPredicate the {@link FlushPredicate}.
+	 * @since 4.3
+	 */
+	public synchronized void flushIfNeeded(FlushPredicate flushPredicate) {
+		Iterator<Entry<String, FileState>> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator();
+		while (iterator.hasNext()) {
+			Entry<String, FileState> entry = iterator.next();
+			FileState state = entry.getValue();
+			if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite)) {
+				iterator.remove();
+				state.close();
+			}
+		}
+	}
+
+	/**
+	 * 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)}
+	 * 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.
+	 * @since 4.3
+	 */
+	public synchronized void flushIfNeeded(MessageFlushPredicate flushPredicate, Message<?> filterMessage) {
+		Iterator<Entry<String, FileState>> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator();
+		while (iterator.hasNext()) {
+			Entry<String, FileState> entry = iterator.next();
+			FileState state = entry.getValue();
+			if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite, filterMessage)) {
+				iterator.remove();
+				state.close();
+			}
+		}
+	}
+
+
+	private static final class FileState {
+
+		private final BufferedWriter writer;
+
+		private final BufferedOutputStream  stream;
+
+		private volatile long lastWrite;
+
+		private FileState(BufferedWriter writer) {
+			this.writer = writer;
+			this.stream = null;
+		}
+
+		private FileState(BufferedOutputStream stream) {
+			this.writer = null;
+			this.stream = stream;
+		}
+
+		private void close() {
+			try {
+				if (this.writer != null) {
+					this.writer.close();
+				}
+				else {
+					this.stream.close();
+				}
+			}
+			catch (IOException e) {
+				;
+			}
+		}
+	}
+
+	private final class Flusher implements Runnable {
+
+		@Override
+		public void run() {
+			synchronized (FileWritingMessageHandler.this) {
+				long expired = FileWritingMessageHandler.this.flushTask == null ? Long.MAX_VALUE
+						: (System.currentTimeMillis() - FileWritingMessageHandler.this.flushInterval);
+				Iterator<Entry<String, FileState>> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator();
+				while (iterator.hasNext()) {
+					Entry<String, FileState> entry = iterator.next();
+					FileState state = entry.getValue();
+					if (state.lastWrite < expired) {
+						iterator.remove();
+						state.close();
+						if (logger.isDebugEnabled()) {
+							logger.debug("Flushed: " + entry.getKey());
+						}
+					}
+				}
+			}
+		}
+
+	}
+
+	/**
+	 * 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.
+	 * @since 4.3
+	 *
+	 */
+	public interface FlushPredicate {
+
+		/**
+		 * @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.
+		 */
+		boolean shouldFlush(String fileAbsolutePath, 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.
+	 * @see FileWritingMessageHandler#trigger(Message)
+	 * @since 4.3
+	 *
+	 */
+	public interface MessageFlushPredicate {
+
+		/**
+		 * @param fileAbsolutePath the path to the 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.
+		 */
+		boolean shouldFlush(String fileAbsolutePath, long lastWrite, Message<?> filterMessage);
+
+	}
+
+	/**
+	 * Flushes files where the path matches a pattern, regardless of last write time.
+	 */
+	private final class DefaultFlushPredicate implements MessageFlushPredicate {
+
+		@Override
+		public boolean shouldFlush(String fileAbsolutePath, long lastWrite, Message<?> triggerMessage) {
+			Pattern pattern;
+			if (triggerMessage.getPayload() instanceof String) {
+				pattern = Pattern.compile((String) triggerMessage.getPayload());
+			}
+			else if (triggerMessage.getPayload() instanceof Pattern) {
+				pattern = (Pattern) triggerMessage.getPayload();
+			}
+			else {
+				throw new IllegalArgumentException("Invalid payload type, must be a String or Pattern");
+			}
+			return pattern.matcher(fileAbsolutePath).matches();
+		}
+
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -34,6 +34,7 @@ import org.springframework.util.StringUtils;
  * @author Artem Bilan
  * @author Gunnar Hillert
  * @author Tony Falabella
+ * @author Gary Russell
  *
  * @since 1.0.3
  */
@@ -41,7 +42,7 @@ abstract class FileWritingMessageHandlerBeanDefinitionBuilder {
 
 	static BeanDefinitionBuilder configure(Element element, boolean expectReply, ParserContext parserContext) {
 
-		BeanDefinitionBuilder builder = 
+		BeanDefinitionBuilder builder =
 				BeanDefinitionBuilder.genericBeanDefinition(FileWritingMessageHandlerFactoryBean.class);
 
 		String directory = element.getAttribute("directory");
@@ -56,7 +57,7 @@ abstract class FileWritingMessageHandlerBeanDefinitionBuilder {
 		}
 
 		if (StringUtils.hasText(directoryExpression)) {
-			BeanDefinitionBuilder expressionBuilder = 
+			BeanDefinitionBuilder expressionBuilder =
 					BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
 			expressionBuilder.addConstructorArgValue(directoryExpression);
 			builder.addPropertyValue("directoryExpression", expressionBuilder.getBeanDefinition());
@@ -70,6 +71,9 @@ abstract class FileWritingMessageHandlerBeanDefinitionBuilder {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "temporary-file-suffix");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "mode", "fileExistsMode");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "charset");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "buffer-size");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "flush-interval");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "flush-predicate");
 		String remoteFileNameGenerator = element.getAttribute("filename-generator");
 		String remoteFileNameGeneratorExpression = element.getAttribute("filename-generator-expression");
 		boolean hasRemoteFileNameGenerator = StringUtils.hasText(remoteFileNameGenerator);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -22,6 +22,7 @@ import org.springframework.expression.Expression;
 import org.springframework.integration.config.AbstractSimpleMessageHandlerFactoryBean;
 import org.springframework.integration.file.FileNameGenerator;
 import org.springframework.integration.file.FileWritingMessageHandler;
+import org.springframework.integration.file.FileWritingMessageHandler.MessageFlushPredicate;
 import org.springframework.integration.file.support.FileExistsMode;
 
 /**
@@ -37,7 +38,7 @@ import org.springframework.integration.file.support.FileExistsMode;
  *
  * @since 1.0.3
  */
-public class FileWritingMessageHandlerFactoryBean 
+public class FileWritingMessageHandlerFactoryBean
 		extends AbstractSimpleMessageHandlerFactoryBean<FileWritingMessageHandler>{
 
 	private volatile File directory;
@@ -61,9 +62,15 @@ public class FileWritingMessageHandlerFactoryBean
 	private volatile FileExistsMode fileExistsMode;
 
 	private volatile boolean expectReply = true;
-	
+
+	private Integer bufferSize;
+
 	private volatile Boolean appendNewLine;
 
+	private volatile Long flushInterval;
+
+	private volatile MessageFlushPredicate flushPredicate;
+
 	public void setFileExistsMode(String fileExistsModeAsString) {
 		this.fileExistsMode = FileExistsMode.getForString(fileExistsModeAsString);
 	}
@@ -111,7 +118,19 @@ public class FileWritingMessageHandlerFactoryBean
 	public void setAppendNewLine(Boolean appendNewLine) {
 		this.appendNewLine = appendNewLine;
 	}
-	
+
+	public void setBufferSize(Integer bufferSize) {
+		this.bufferSize = bufferSize;
+	}
+
+	public void setFlushInterval(long flushInterval) {
+		this.flushInterval = flushInterval;
+	}
+
+	public void setFlushPredicate(MessageFlushPredicate flushPredicate) {
+		this.flushPredicate = flushPredicate;
+	}
+
 	@Override
 	protected FileWritingMessageHandler createHandler() {
 
@@ -157,8 +176,17 @@ public class FileWritingMessageHandlerFactoryBean
 		if (this.fileExistsMode != null) {
 			handler.setFileExistsMode(this.fileExistsMode);
 		}
+		if (this.bufferSize != null) {
+			handler.setBufferSize(this.bufferSize);
+		}
+		if (this.flushInterval != null) {
+			handler.setFlushInterval(this.flushInterval);
+		}
+		if (this.flushPredicate != null) {
+			handler.setFlushPredicate(this.flushPredicate);
+		}
 
 		return handler;
 	}
-	
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/support/FileExistsMode.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -23,16 +23,24 @@ import org.springframework.util.StringUtils;
  * case the destination file already exists.
  *
  * @author Gunnar Hillert
+ * @author Gary Russell
  * @since 2.2
  *
  */
 public enum FileExistsMode {
 
 	/**
-	 * Append data to any pre-existing files.
+	 * Append data to any pre-existing files; close after each append.
 	 */
 	APPEND,
 
+	/**
+	 * Append data to any pre-existing files; do not flush/close after
+	 * appending.
+	 * @since 4.3
+	 */
+	APPEND_NO_FLUSH,
+
 	/**
 	 * Raise an exception in case the file to be written already exists.
 	 */

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

@@ -466,6 +466,40 @@ Only files matching this regular expression will be picked up by this adapter.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="buffer-size" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					The buffer size to use when writing to files.
+					Default 8192 bytes.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="flush-interval" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					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-predicate" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					When using 'mode=APPEND_NO_FLUSH',
+					a reference to a 'FlushPredicate' implementation used when a message is
+					sent to the message handler's 'MessageTriggerAction.trigger()' method.
+					By default, the payload of such a message must be a Regex used to match
+					the file absolute path.
+				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type
+							type="org.springframework.integration.file.FileWritingMessageHandler.MessageFlushPredicate"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
     </xsd:complexType>
 
@@ -695,6 +729,19 @@ Only files matching this regular expression will be picked up by this adapter.
 					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:enumeration>
+			<xsd:enumeration value="APPEND_NO_FLUSH">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Same as 'APPEND' but the data is not flushed or the file
+						closed. This can significantly improve performance at the
+						risk of lost data in the event of a failure.
+
+						Various strategies are available for flushing the data and
+						closing the file. Refer to the reference documentation for
+						more information.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:enumeration>
 			<xsd:enumeration value="FAIL">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
@@ -813,13 +860,23 @@ Only files matching this regular expression will be picked up by this adapter.
 					subsequent data will be appended to it. This attribute
 					is mutualy exclusive with the use of a temporary file,
 					since append is done to the actual file and not its
-					temporary counterpart.
+					temporary counterpart. The file is closed after each write.
 
 					If set to APPEND, the component will also use
 					instance of the LockRegistry to ensure that there are no
 					collisions when multiple threads are writing to the same
 					file.
 
+					APPEND_NO_FLUSH:
+
+					Same as 'APPEND' but the data is not flushed or the file
+					closed. This can significantly improve performance at the
+					risk of lost data in the event of a failure.
+
+					Various strategies are available for flushing the data and
+					closing the file. Refer to the reference documentation for
+					more information.
+
 					FAIL:
 
 					If the target file exists, a MessageHandlingException

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

@@ -21,6 +21,9 @@ import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertThat;
@@ -28,11 +31,15 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
 
+import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
 
 import org.junit.Before;
 import org.junit.Ignore;
@@ -43,12 +50,15 @@ import org.junit.rules.TemporaryFolder;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.file.FileWritingMessageHandler.FlushPredicate;
+import org.springframework.integration.file.FileWritingMessageHandler.MessageFlushPredicate;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.FileCopyUtils;
 
 /**
@@ -425,6 +435,78 @@ public class FileWritingMessageHandlerTests {
 		assertFileContentIs(outFile, "foo");
 	}
 
+	@Test
+	public void noFlushAppend() throws Exception {
+		File tempFolder = this.temp.newFolder();
+		FileWritingMessageHandler handler = new FileWritingMessageHandler(tempFolder);
+		handler.setFileExistsMode(FileExistsMode.APPEND_NO_FLUSH);
+		handler.setFileNameGenerator(new FileNameGenerator() {
+
+			@Override
+			public String generateFileName(Message<?> message) {
+				return "foo.txt";
+			}
+		});
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		handler.setTaskScheduler(taskScheduler);
+		handler.setOutputChannel(new NullChannel());
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setFlushInterval(30000);
+		handler.afterPropertiesSet();
+		handler.start();
+		File file = new File(tempFolder, "foo.txt");
+		handler.handleMessage(new GenericMessage<String>("foo"));
+		handler.handleMessage(new GenericMessage<String>("bar"));
+		handler.handleMessage(new GenericMessage<String>("baz"));
+		handler.handleMessage(new GenericMessage<byte[]>("qux".getBytes())); // change of payload type forces flush
+		assertThat(file.length(), greaterThanOrEqualTo(9L));
+		handler.stop(); // forces flush
+		assertThat(file.length(), equalTo(12L));
+		handler.setFlushInterval(100);
+		handler.start();
+		handler.handleMessage(new GenericMessage<InputStream>(new ByteArrayInputStream("fiz".getBytes())));
+		int n = 0;
+		while (n++ < 100 && file.length() < 15) {
+			Thread.sleep(100);
+		}
+		assertThat(file.length(), equalTo(15L));
+		handler.handleMessage(new GenericMessage<InputStream>(new ByteArrayInputStream("buz".getBytes())));
+		handler.trigger(new GenericMessage<String>(Matcher.quoteReplacement(file.getAbsolutePath())));
+		assertThat(file.length(), equalTo(18L));
+		assertEquals(0, TestUtils.getPropertyValue(handler, "fileStates", Map.class).size());
+
+		handler.setFlushInterval(30000);
+		final AtomicBoolean called = new AtomicBoolean();
+		handler.setFlushPredicate(new MessageFlushPredicate() {
+
+			@Override
+			public boolean shouldFlush(String fileAbsolutePath, long lastWrite, Message<?> triggerMessage) {
+				called.set(true);
+				return true;
+			}
+
+		});
+		handler.handleMessage(new GenericMessage<InputStream>(new ByteArrayInputStream("box".getBytes())));
+		handler.trigger(new GenericMessage<String>("foo"));
+		assertThat(file.length(), equalTo(21L));
+		assertTrue(called.get());
+
+		handler.handleMessage(new GenericMessage<InputStream>(new ByteArrayInputStream("bux".getBytes())));
+		called.set(false);
+		handler.flushIfNeeded(new FlushPredicate() {
+
+			@Override
+			public boolean shouldFlush(String fileAbsolutePath, long lastWrite) {
+				called.set(true);
+				return true;
+			}
+
+		});
+		assertThat(file.length(), equalTo(24L));
+		assertTrue(called.get());
+	}
+
 	void assertFileContentIsMatching(Message<?> result) throws IOException {
 		assertFileContentIs(result, SAMPLE_CONTENT);
 	}

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

@@ -49,6 +49,20 @@
 								   auto-startup="false"
 								   directory="${java.io.tmpdir}"/>
 
+	<file:outbound-channel-adapter id="adapterWithFlushing"
+								   channel="testChannel"
+								   order="555"
+								   mode="APPEND_NO_FLUSH"
+								   flush-interval="12345"
+								   flush-predicate="predicate"
+								   buffer-size="4096"
+								   auto-startup="false"
+								   directory="${java.io.tmpdir}"/>
+
+	<bean id="predicate" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.integration.file.FileWritingMessageHandler.MessageFlushPredicate" />
+	</bean>
+
 	<file:outbound-channel-adapter id="usageChannel"
 								   filename-generator-expression="@fooString"
 								   mode="APPEND"

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.file.config;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
@@ -38,12 +39,15 @@ import org.springframework.expression.Expression;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.file.DefaultFileNameGenerator;
 import org.springframework.integration.file.FileWritingMessageHandler;
+import org.springframework.integration.file.FileWritingMessageHandler.MessageFlushPredicate;
+import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.FileCopyUtils;
@@ -62,6 +66,7 @@ import org.springframework.util.ReflectionUtils;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class FileOutboundChannelAdapterParserTests {
 
 	@Autowired
@@ -76,6 +81,9 @@ public class FileOutboundChannelAdapterParserTests {
 	@Autowired
 	EventDrivenConsumer adapterWithOrder;
 
+	@Autowired
+	EventDrivenConsumer adapterWithFlushing;
+
 	@Autowired
 	EventDrivenConsumer adapterWithCharset;
 
@@ -106,6 +114,9 @@ public class FileOutboundChannelAdapterParserTests {
 	@Autowired
 	CountDownLatch fileWriteLatch;
 
+	@Autowired
+	MessageFlushPredicate predicate;
+
 	private volatile static int adviceCalled;
 
 	@Test
@@ -165,6 +176,18 @@ public class FileOutboundChannelAdapterParserTests {
 		assertEquals(555, handlerAccessor.getPropertyValue("order"));
 	}
 
+	@Test
+	public void adapterWithFlushing() {
+		DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapterWithFlushing);
+		FileWritingMessageHandler handler = (FileWritingMessageHandler)
+				adapterAccessor.getPropertyValue("handler");
+		DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler);
+		assertEquals(4096, handlerAccessor.getPropertyValue("bufferSize"));
+		assertEquals(12345L, handlerAccessor.getPropertyValue("flushInterval"));
+		assertEquals(FileExistsMode.APPEND_NO_FLUSH, handlerAccessor.getPropertyValue("fileExistsMode"));
+		assertSame(this.predicate, handlerAccessor.getPropertyValue("flushPredicate"));
+	}
+
 	@Test
 	public void adapterWithAutoStartupFalse() {
 		DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapterWithOrder);

src/reference/asciidoc/file.adoc

@@ -438,7 +438,10 @@ This class can deal with the following payload types:
 
 You can configure the encoding and the charset that will be used in case of a String payload.
 
-To make things easier, you can configure the `FileWritingMessageHandler` as part of an _Outbound Channel Adapter_ or _Outbound Gateway_ using the provided XML namespace support.
+To make things easier, you can configure the `FileWritingMessageHandler` as part of an _Outbound Channel Adapter_ or
+_Outbound Gateway_ using the provided XML namespace support.
+
+Starting with _version 4.3_, you can specify the buffer size to use when writing files.
 
 [[file-writing-file-names]]
 ==== Generating File Names
@@ -524,6 +527,7 @@ The following options exist:
 
 * REPLACE (Default)
 * APPEND
+* APPEND_NO_FLUSH
 * FAIL
 * IGNORE
 
@@ -539,7 +543,15 @@ If the _mode_ attribute is not specified, then this is the default behavior when
 _APPEND_
 
 This mode allows you to append Message content to the existing file instead of creating a new file each time.
-Note that this attribute is mutually exclusive with _temporary-file-suffix_ attribute since when appending content to the existing file, the adapter no longer uses a temporary file.
+Note that this attribute is mutually exclusive with _temporary-file-suffix_ attribute since when appending content to
+the existing file, the adapter no longer uses a temporary file.
+The file is closed after each message.
+
+__APPEND_NO_FLUSH__
+
+This has the same semantics as *APPEND* but the data is not flushed and the file is not closed after each message.
+This can provide a significant performance at the risk of data loss in the case of a failure.
+See <<file-flushing>> for more information.
 
 _FAIL_
 
@@ -551,6 +563,27 @@ If the target file exists, the message payload is silently ignored.
 
 NOTE: When using a temporary file suffix (default: `.writing`), the _IGNORE_ mode will apply if the final file name exists, or the temporary file name exists.
 
+[[file-flushing]]
+==== Flushing Files When using APPEND_NO_FLUSH
+
+The *APPEND_NO_FLUSH* mode was added in _version 4.3_.
+This can improve performance because the file is not closed after each message.
+However, this can cause data loss in the event of a failure.
+
+Several flushing strategies, to mitigate this data loss, are provided:
+
+- `flushInterval` - if a file is not written to for this period of time, it is automatically flushed.
+This is approximate and may be up to `1.33x` this time.
+- Send a message to the message handler's `trigger` method containing a regular expression.
+Files with absolute path names matching the pattern will be flushed.
+- Provide the handler with a custom `MessageFlushPredicate` implementation to modify the action taken when a message
+is sent to the `trigger` method.
+- Invoke one of the handler's `flushIfNeeded` methods passing in a custom `FileWritingMessageHandler.FlushPredicate`
+or `FileWritingMessageHandler.MessageFlushPredicate` implementation.
+
+The predicates are called for each open file.
+See the java docs for these interfaces for more information.
+
 [[file-outbound-channel-adapter]]
 ==== File Outbound Channel Adapter
 

src/reference/asciidoc/message-store.adoc

@@ -86,7 +86,7 @@ For this reason, users should not perform such manipulation, or set the `copyOnG
 =====
 
 [[message-group-factory]]
-===== MessageGroupFactory
+==== MessageGroupFactory
 
 Starting with _version 4.3_, some `MessageGroupStore` implementations can be injected with a custom
 `MessageGroupFactory` strategy to create/customize the `MessageGroup` instances used by the `MessageGroupStore`.

src/reference/asciidoc/whats-new.adoc

@@ -57,10 +57,21 @@ See <<udp-adapters>> for more information.
 
 ==== File Changes
 
+===== Destination Directory Creation
+
 The generated file name for the `FileWritingMessageHandler` can represent _sub-path_ to save the desired directory
 structure for file in the target directory.
 See <<file-writing-file-names>> for more information.
 
+===== Buffer Size
+
+When writing files, you can now specify the buffer size to use.
+
+===== Appending and Flushing
+
+You can now avoid flushing files when appending and use a number of strategies to flush the data during idle periods.
+See <<file-flushing>> for more information.
+
 ==== AMQP Changes
 
 The outbound endpoints now support a `RabbitTemplate` configured with a `ContentTypeDelegatingMessageConverter` such