Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4502: Add new file callback in FileWritingMH

Browse Source 
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`
This commit is contained in:
Alen Turkovic
2018-07-23 10:44:21 +02:00
committed by Artem Bilan
parent a7ffbde07f
commit 9ac0746f19
4 changed files with 87 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<PosixFilePermission> permissions;
private BiConsumer<File, Message<?>> 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<File, Message<?>> 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 {

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

@@ -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);
}

4
src/reference/asciidoc/file.adoc
View File

@@ -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<File, Message<?>>` `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

7
src/reference/asciidoc/whats-new.adoc
View File

@@ -86,6 +86,13 @@ Starting with version 5.1, you must explicitly turn on the `@Publisher` AOP func
See "`<<publisher-annotation>>`" 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