From 9ac0746f1940ff3b9324c6ebc53b9089563ed356 Mon Sep 17 00:00:00 2001 From: Alen Turkovic Date: Mon, 23 Jul 2018 10:44:21 +0200 Subject: [PATCH] INT-4502: Add new file callback in FileWritingMH JIRA: https://jira.spring.io/browse/INT-4502 Documentation Using a BiConsumer instead of a custom interface Doc fixes Calling callback while locked More requested fixes Missing `FileWritingMessageHandler.this` on `fileExistsMode` --- .../file/FileWritingMessageHandler.java | 58 +++++++++++++++---- .../file/FileWritingMessageHandlerTests.java | 31 +++++++++- src/reference/asciidoc/file.adoc | 4 ++ src/reference/asciidoc/whats-new.adoc | 7 +++ 4 files changed, 87 insertions(+), 13 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 5b066d5fbe..2e9617e7e0 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -39,6 +39,7 @@ import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.Lock; +import java.util.function.BiConsumer; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -105,6 +106,7 @@ import org.springframework.util.StringUtils; * @author Gunnar Hillert * @author Gary Russell * @author Tony Falabella + * @author Alen Turkovic */ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHandler implements Lifecycle, MessageTriggerAction { @@ -159,6 +161,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand private Set permissions; + private BiConsumer> newFileCallback; + /** * Constructor which sets the {@link #destinationDirectoryExpression} using * a {@link LiteralExpression}. @@ -443,6 +447,18 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand this.permissions = permissions; } + /** + * Set the callback to use when creating new files. This callback will only be called + * if {@link #fileExistsMode} is {@link FileExistsMode#APPEND} or {@link FileExistsMode#APPEND_NO_FLUSH} + * and new file has to be created. The callback receives the new result file and the message that + * triggered the handler. + * @param newFileCallback callback + * @since 5.1 + */ + public void setNewFileCallback(final BiConsumer> newFileCallback) { + this.newFileCallback = newFileCallback; + } + @Override protected void doInit() { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); @@ -555,20 +571,21 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand .contains(File.separator)) { resultFile.getParentFile().mkdirs(); //NOSONAR - will fail on the writing below } + if (payload instanceof File) { - resultFile = handleFileMessage((File) payload, tempFile, resultFile); + resultFile = handleFileMessage((File) payload, tempFile, resultFile, requestMessage); } else if (payload instanceof InputStream) { resultFile = handleInputStreamMessage((InputStream) payload, originalFileFromHeader, tempFile, - resultFile); + resultFile, requestMessage); } else if (payload instanceof byte[]) { resultFile = this.handleByteArrayMessage( - (byte[]) payload, originalFileFromHeader, tempFile, resultFile); + (byte[]) payload, originalFileFromHeader, tempFile, resultFile, requestMessage); } else if (payload instanceof String) { resultFile = this.handleStringMessage( - (String) payload, originalFileFromHeader, tempFile, resultFile); + (String) payload, originalFileFromHeader, tempFile, resultFile, requestMessage); } else { throw new IllegalArgumentException( @@ -621,19 +638,20 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand return null; } - private File handleFileMessage(final File sourceFile, File tempFile, final File resultFile) throws IOException { + private File handleFileMessage(final File sourceFile, File tempFile, final File resultFile, final Message requestMessage) + throws IOException { if (!FileExistsMode.APPEND.equals(this.fileExistsMode) && this.deleteSourceFiles) { rename(sourceFile, resultFile); return resultFile; } else { BufferedInputStream bis = new BufferedInputStream(new FileInputStream(sourceFile)); - return handleInputStreamMessage(bis, sourceFile, tempFile, resultFile); + return handleInputStreamMessage(bis, sourceFile, tempFile, resultFile, requestMessage); } } private File handleInputStreamMessage(final InputStream sourceFileInputStream, File originalFile, File tempFile, - final File resultFile) throws IOException { + final File resultFile, final Message requestMessage) throws IOException { final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode) || FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode); @@ -645,6 +663,10 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand @Override protected void whileLocked() throws IOException { + if (FileWritingMessageHandler.this.newFileCallback != null && !fileToWriteTo.exists()) { + FileWritingMessageHandler.this.newFileCallback.accept(fileToWriteTo, requestMessage); + } + FileState state = getFileState(fileToWriteTo, false); BufferedOutputStream bos = null; try { @@ -719,8 +741,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand } } - private File handleByteArrayMessage(final byte[] bytes, File originalFile, File tempFile, final File resultFile) - throws IOException { + private File handleByteArrayMessage(final byte[] bytes, File originalFile, File tempFile, final File resultFile, + final Message requestMessage) throws IOException { final File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile); final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode); @@ -730,6 +752,12 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand @Override protected void whileLocked() throws IOException { + if (FileWritingMessageHandler.this.newFileCallback != null && !fileToWriteTo.exists() && + (FileExistsMode.APPEND.equals(FileWritingMessageHandler.this.fileExistsMode) + || FileExistsMode.APPEND_NO_FLUSH.equals(FileWritingMessageHandler.this.fileExistsMode))) { + FileWritingMessageHandler.this.newFileCallback.accept(fileToWriteTo, requestMessage); + } + FileState state = getFileState(fileToWriteTo, false); BufferedOutputStream bos = null; try { @@ -762,8 +790,8 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand return resultFile; } - private File handleStringMessage(final String content, File originalFile, File tempFile, final File resultFile) - throws IOException { + private File handleStringMessage(final String content, File originalFile, File tempFile, final File resultFile, + final Message requestMessage) throws IOException { final File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile); final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode); @@ -773,6 +801,12 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand @Override protected void whileLocked() throws IOException { + if (FileWritingMessageHandler.this.newFileCallback != null && !fileToWriteTo.exists() && + (FileExistsMode.APPEND.equals(FileWritingMessageHandler.this.fileExistsMode) + || FileExistsMode.APPEND_NO_FLUSH.equals(FileWritingMessageHandler.this.fileExistsMode))) { + FileWritingMessageHandler.this.newFileCallback.accept(fileToWriteTo, requestMessage); + } + FileState state = getFileState(fileToWriteTo, true); BufferedWriter writer = null; try { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java index 68a59d83d1..f3347df2f1 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -85,6 +85,7 @@ import org.springframework.util.FileCopyUtils; * @author Tony Falabella * @author Gunnar Hillert * @author Artem Bilan + * @author Alen Turkovic */ public class FileWritingMessageHandlerTests { @@ -657,6 +658,34 @@ public class FileWritingMessageHandlerTests { assertLastModifiedIs(result, 43_000_000); } + @Test + public void newFileCallback() throws Exception { + long lastModified = 1234000L; + Message message = MessageBuilder.withPayload("bar") + .setHeader(FileHeaders.SET_MODIFIED, lastModified) + .build(); + QueueChannel output = new QueueChannel(); + handler.setFileExistsMode(FileExistsMode.APPEND); + handler.setCharset(DEFAULT_ENCODING); + handler.setOutputChannel(output); + handler.setPreserveTimestamp(true); + handler.setNewFileCallback((file, msg) -> { + try { + FileCopyUtils.copy(("foo" + System.lineSeparator()).getBytes(DEFAULT_ENCODING), + new FileOutputStream(file, false)); + } + catch (IOException e) { + fail("unexpected copy exception"); + } + }); + handler.handleMessage(message); + Message result = output.receive(0); + assertFileContentIs(result, "foo" + System.lineSeparator() + "bar"); + assertLastModifiedIs(result, lastModified); + handler.handleRequestMessage(message); + assertFileContentIs(result, "foo" + System.lineSeparator() + "barbar"); + } + void assertFileContentIsMatching(Message result) throws IOException { assertFileContentIs(result, SAMPLE_CONTENT); } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index a715f1b2bd..26fa1ff9e6 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -616,6 +616,10 @@ outbound gateway by using the XML namespace. Starting with version 4.3, you can specify the buffer size to use when writing files. +Starting with version 5.1, you can provide a `BiConsumer>` `newFileCallback` which will be triggered if you are using `FileExistsMode.APPEND` or `FileExistsMode.APPEND_NO_FLUSH` and a new file has to be created. +This callback receives the newly created file and the message which triggered it. +This callback could be used to write a CSV header defined in the message header, for an example. + [[file-writing-file-names]] ==== Generating File Names diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 9d645eb20c..2b37e68cf8 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -86,6 +86,13 @@ Starting with version 5.1, you must explicitly turn on the `@Publisher` AOP func See "`<>`" for more information. +[[x5.1-file-writing-message-handler]] +=== `FileWritingMessageHandler`: New file callback + +If you are using `FileExistsMode.APPEND` or `FileExistsMode.APPEND_NO_FLUSH` you can provide a `newFileCallback` that will be called when creating a new file. +This callback receives the newly created file and the message that triggered the callback. +This could be used to write a CSV header, for an example. + [[x5.1-amqp]] === AMQP Changes