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