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 7ef15b4309..ef1813dceb 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 @@ -365,13 +365,13 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); if (this.destinationDirectoryExpression instanceof LiteralExpression) { - final File directory = new File(this.destinationDirectoryExpression.getValue( - this.evaluationContext, null, String.class)); + String directoryPath = this.destinationDirectoryExpression.getValue(this.evaluationContext, String.class); + final File directory = new File(directoryPath); validateDestinationDirectory(directory, this.autoCreateDirectory); } Assert.state(!(this.temporaryFileSuffixSet - && (FileExistsMode.APPEND.equals(this.fileExistsMode) + && (FileExistsMode.APPEND.equals(this.fileExistsMode) || FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode))), "'temporaryFileSuffix' can not be set when appending to an existing file"); @@ -391,12 +391,32 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } @Override - public synchronized void stop() { - if (this.flushTask != null) { - this.flushTask.cancel(true); - this.flushTask = null; + public void stop() { + synchronized (this) { + if (this.flushTask != null) { + this.flushTask.cancel(true); + this.flushTask = null; + } + } + Flusher flusher = new Flusher(); + flusher.run(); + boolean needInterrupt = this.fileStates.size() > 0; + int n = 0; + while (n++ < 10 && this.fileStates.size() > 0) { + try { + Thread.sleep(1); + } + catch (InterruptedException e) { + // cancel the interrupt + } + flusher.run(); + } + if (this.fileStates.size() > 0) { + this.logger.error("Failed to flush after multiple attempts, while stopping: " + this.fileStates.keySet()); + } + if (needInterrupt) { + Thread.currentThread().interrupt(); } - new Flusher().run(); } @Override @@ -529,7 +549,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } private File handleInputStreamMessage(final InputStream sourceFileInputStream, File originalFile, File tempFile, - final File resultFile) throws IOException { + final File resultFile) throws IOException { final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode) || FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode); @@ -796,10 +816,10 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand 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) { + String absolutePath = fileToWriteTo.getAbsolutePath(); state = this.fileStates.get(absolutePath); if (state != null && ((isString && state.stream != null) || (!isString && state.writer != null))) { state.close(); @@ -872,16 +892,19 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand * @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(); - } - } + public void flushIfNeeded(final FlushPredicate flushPredicate) { + flushIfNeeded(new MessageFlushPredicate() { + + @Override + public boolean shouldFlush(String fileAbsolutePath, long lastWrite, + Message filterMessage) { + + return flushPredicate.shouldFlush(fileAbsolutePath, lastWrite); + + } + + }, + null); } /** @@ -893,16 +916,24 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand * @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(); + public void flushIfNeeded(MessageFlushPredicate flushPredicate, Message filterMessage) { + doFlush(findFilesToFlush(flushPredicate, filterMessage)); + } + + private Map findFilesToFlush(MessageFlushPredicate flushPredicate, Message filterMessage) { + Map toRemove = new HashMap(); + synchronized (this) { + Iterator> iterator = 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(); + toRemove.put(entry.getKey(), state); + } } } + return toRemove; } private synchronized void clearState(final File fileToWriteTo, final FileState state) { @@ -911,6 +942,33 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } } + private void doFlush(Map toRemove) { + Map toRestore = new HashMap(); + boolean interrupted = false; + for (Entry entry : toRemove.entrySet()) { + if (!interrupted && entry.getValue().close()) { + if (FileWritingMessageHandler.this.logger.isDebugEnabled()) { + FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey()); + } + } + else { // interrupted (stop), re-add + interrupted = true; + toRestore.put(entry.getKey(), entry.getValue()); + } + } + if (interrupted) { + if (FileWritingMessageHandler.this.logger.isDebugEnabled()) { + FileWritingMessageHandler.this.logger + .debug("Interrupted during flush; not flushed: " + toRestore.keySet()); + } + synchronized (this) { + for (Entry entry : toRestore.entrySet()) { + this.fileStates.putIfAbsent(entry.getKey(), entry.getValue()); + } + } + } + } + private static boolean rename(File source, File target) throws IOException { return (nioFilesPresent && filesMove(source, target)) || source.renameTo(target); } @@ -925,7 +983,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand private final BufferedWriter writer; - private final BufferedOutputStream stream; + private final BufferedOutputStream stream; private final Lock lock; @@ -975,6 +1033,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand @Override public void run() { + Map toRemove = new HashMap(); synchronized (FileWritingMessageHandler.this) { long expired = FileWritingMessageHandler.this.flushTask == null ? Long.MAX_VALUE : (System.currentTimeMillis() - FileWritingMessageHandler.this.flushInterval); @@ -984,18 +1043,12 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand FileState state = entry.getValue(); if (state.lastWrite < expired || (!FileWritingMessageHandler.this.flushWhenIdle && state.firstWrite < expired)) { - if (state.close()) { - if (FileWritingMessageHandler.this.logger.isDebugEnabled()) { - FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey()); - } - iterator.remove(); - } - else { - break; // interrupted - } + toRemove.put(entry.getKey(), state); + iterator.remove(); } } } + doFlush(toRemove); } }