From c7724c340cb19eb364081cbd107802e719dd3f5f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 12 Jan 2016 16:22:43 -0500 Subject: [PATCH] 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. --- .../file/FileWritingMessageHandler.java | 377 +++++++++++++++++- ...ngMessageHandlerBeanDefinitionBuilder.java | 10 +- .../FileWritingMessageHandlerFactoryBean.java | 38 +- .../file/support/FileExistsMode.java | 12 +- .../config/spring-integration-file-4.3.xsd | 59 ++- .../file/FileWritingMessageHandlerTests.java | 82 ++++ ...boundChannelAdapterParserTests-context.xml | 14 + ...FileOutboundChannelAdapterParserTests.java | 25 +- src/reference/asciidoc/file.adoc | 37 +- src/reference/asciidoc/message-store.adoc | 2 +- src/reference/asciidoc/whats-new.adoc | 11 + 11 files changed, 635 insertions(+), 32 deletions(-) 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 c7cfb70b1d..40f56bbc45 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 @@ -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 * toString() method by the * {@link org.springframework.integration.transformer.ObjectToStringTransformer}. + *

+ * {@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 fileStates = new HashMap(); + 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> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator(); + while (iterator.hasNext()) { + Entry 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> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator(); + while (iterator.hasNext()) { + Entry 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> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator(); + while (iterator.hasNext()) { + Entry 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(); + } + + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java index 771385d20d..96f965bd9c 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerBeanDefinitionBuilder.java +++ b/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); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java index f125e89ae5..d216e10f3b 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileWritingMessageHandlerFactoryBean.java +++ b/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{ 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; } - + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/support/FileExistsMode.java b/spring-integration-file/src/main/java/org/springframework/integration/file/support/FileExistsMode.java index fd04f0713e..869884f6d8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/support/FileExistsMode.java +++ b/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. */ diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd index b611a1cb47..0a4ee677ad 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd +++ b/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. + + + + The buffer size to use when writing to files. + Default 8192 bytes. + + + + + + + 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. + + + + + + + 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. + + + + + + + + @@ -695,6 +729,19 @@ Only files matching this regular expression will be picked up by this adapter. ]]> + + + + + 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("foo")); + handler.handleMessage(new GenericMessage("bar")); + handler.handleMessage(new GenericMessage("baz")); + handler.handleMessage(new GenericMessage("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(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(new ByteArrayInputStream("buz".getBytes()))); + handler.trigger(new GenericMessage(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(new ByteArrayInputStream("box".getBytes()))); + handler.trigger(new GenericMessage("foo")); + assertThat(file.length(), equalTo(21L)); + assertTrue(called.get()); + + handler.handleMessage(new GenericMessage(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); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml index 72d130e294..abdcbcfd97 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundChannelAdapterParserTests-context.xml +++ b/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}"/> + + + + + + > 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 diff --git a/src/reference/asciidoc/message-store.adoc b/src/reference/asciidoc/message-store.adoc index 5045aa4e6d..8c64f102fe 100644 --- a/src/reference/asciidoc/message-store.adoc +++ b/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`. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 6410c5ffc9..c13c088ed1 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -57,10 +57,21 @@ See <> 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 <> 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 <> for more information. + ==== AMQP Changes The outbound endpoints now support a `RabbitTemplate` configured with a `ContentTypeDelegatingMessageConverter` such