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

Polishing for the FileWritingMessageHandler

Browse Source 
* Fix typos in docs
* Add `om.google.code.findbugs:jsr305` to avoid warnings like
`Cannot find annotation`
This commit is contained in:
Artem Bilan
2018-07-24 15:49:22 -04:00
parent 9ac0746f19
commit 44b7df94c1
5 changed files with 81 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
build.gradle
View File

@@ -160,6 +160,9 @@ subprojects { subproject ->
testCompile project(":spring-integration-test-support")
}
// JSR-305 only used for non-required meta-annotations
compileOnly("com.google.code.findbugs:jsr305:3.0.2")
testCompile "org.junit.jupiter:junit-jupiter-api:$junitJupiterVersion"
testRuntime "org.junit.jupiter:junit-jupiter-engine:$junitJupiterVersion"
testRuntime "org.junit.platform:junit-platform-launcher:$junitPlatformVersion"

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

@@ -111,7 +111,7 @@ import org.springframework.util.StringUtils;
public class FileWritingMessageHandler extends AbstractReplyProducingMessageHandler
implements Lifecycle, MessageTriggerAction {
private static final String LINE_SEPARATOR = System.getProperty("line.separator");
private final Log logger = LogFactory.getLog(this.getClass());
private static final int DEFAULT_BUFFER_SIZE = 8192;
@@ -119,54 +119,51 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
private final Map<String, FileState> fileStates = new HashMap<String, FileState>();
private volatile String temporaryFileSuffix = ".writing";
private volatile boolean temporaryFileSuffixSet = false;
private volatile FileExistsMode fileExistsMode = FileExistsMode.REPLACE;
private final Log logger = LogFactory.getLog(this.getClass());
private volatile FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
private volatile boolean fileNameGeneratorSet;
private volatile StandardEvaluationContext evaluationContext;
private final Expression destinationDirectoryExpression;
private volatile boolean autoCreateDirectory = true;
private String temporaryFileSuffix = ".writing";
private volatile boolean deleteSourceFiles;
private boolean temporaryFileSuffixSet = false;
private volatile Charset charset = Charset.defaultCharset();
private FileExistsMode fileExistsMode = FileExistsMode.REPLACE;
private volatile boolean expectReply = true;
private FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
private volatile boolean appendNewLine = false;
private boolean fileNameGeneratorSet;
private volatile LockRegistry lockRegistry = new PassThruLockRegistry();
private StandardEvaluationContext evaluationContext;
private volatile int bufferSize = DEFAULT_BUFFER_SIZE;
private boolean autoCreateDirectory = true;
private volatile long flushInterval = DEFAULT_FLUSH_INTERVAL;
private boolean deleteSourceFiles;
private volatile boolean flushWhenIdle = true;
private Charset charset = Charset.defaultCharset();
private volatile ScheduledFuture<?> flushTask;
private boolean expectReply = true;
private volatile MessageFlushPredicate flushPredicate = new DefaultFlushPredicate();
private boolean appendNewLine = false;
private volatile boolean preserveTimestamp;
private LockRegistry lockRegistry = new PassThruLockRegistry();
private int bufferSize = DEFAULT_BUFFER_SIZE;
private long flushInterval = DEFAULT_FLUSH_INTERVAL;
private boolean flushWhenIdle = true;
private MessageFlushPredicate flushPredicate = new DefaultFlushPredicate();
private boolean preserveTimestamp;
private Set<PosixFilePermission> permissions;
private BiConsumer<File, Message<?>> newFileCallback;
private volatile ScheduledFuture<?> flushTask;
/**
* Constructor which sets the {@link #destinationDirectoryExpression} using
* a {@link LiteralExpression}.
*
* @param destinationDirectory Must not be null
* @see #FileWritingMessageHandler(Expression)
*/
@@ -177,7 +174,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
/**
* Constructor which sets the {@link #destinationDirectoryExpression}.
*
* @param destinationDirectoryExpression Must not be null
* @see #FileWritingMessageHandler(File)
*/
@@ -192,7 +188,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* <em>true</em>. If set to <em>false</em> and the
* destination directory does not exist, an Exception will be thrown upon
* initialization.
*
* @param autoCreateDirectory true to create the directory if needed.
*/
public void setAutoCreateDirectory(boolean autoCreateDirectory) {
@@ -203,7 +198,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* By default, every file that is in the process of being transferred will
* appear in the file system with an additional suffix, which by default is
* ".writing". This can be changed by setting this property.
*
* @param temporaryFileSuffix The temporary file suffix.
*/
public void setTemporaryFileSuffix(String temporaryFileSuffix) {
@@ -218,15 +212,12 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* case the destination exists. For example {@link FileExistsMode#APPEND}
* instructs this handler to append data to the existing file rather then
* creating a new file for each {@link Message}.
* <p>
* If set to {@link FileExistsMode#APPEND}, the adapter will also
* <p> If set to {@link FileExistsMode#APPEND}, the adapter will also
* create a real instance of the {@link LockRegistry} to ensure that there
* is no collisions when multiple threads are writing to the same file.
* <p>
* Otherwise the LockRegistry is set to {@link PassThruLockRegistry} which
* <p> Otherwise the LockRegistry is set to {@link PassThruLockRegistry} which
* has no effect.
* <p>
* With {@link FileExistsMode#REPLACE_IF_MODIFIED}, if the file exists,
* <p> With {@link FileExistsMode#REPLACE_IF_MODIFIED}, if the file exists,
* it is only replaced if its last modified timestamp is different to the
* source; otherwise, the write is ignored. For {@link File} payloads,
* the actual timestamp of the {@link File} is compared; for other payloads,
@@ -234,12 +225,10 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* If the header is missing, or its value is not a {@link Number}, the file
* is always replaced. This mode will typically only make sense if
* {@link #setPreserveTimestamp(boolean) preserveTimestamp} is true.
*
* @param fileExistsMode Must not be null
* @see #setPreserveTimestamp(boolean)
*/
public void setFileExistsMode(FileExistsMode fileExistsMode) {
Assert.notNull(fileExistsMode, "'fileExistsMode' must not be null.");
this.fileExistsMode = fileExistsMode;
@@ -254,7 +243,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
/**
* Specify whether a reply Message is expected. If not, this handler will simply return null for a
* successful response or throw an Exception for a non-successful response. The default is true.
*
* @param expectReply true if a reply is expected.
*/
public void setExpectReply(boolean expectReply) {
@@ -263,7 +251,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
/**
* If 'true' will append a new-line after each write. It is 'false' by default.
*
* @param appendNewLine true if a new-line should be written to the file after payload is written
* @since 4.0.7
*/
@@ -278,7 +265,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
/**
* Provide the {@link FileNameGenerator} strategy to use when generating
* the destination file's name.
*
* @param fileNameGenerator The file name generator.
*/
public void setFileNameGenerator(FileNameGenerator fileNameGenerator) {
@@ -293,7 +279,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* will only have an effect if the inbound Message has a File payload or
* a {@link FileHeaders#ORIGINAL_FILE} header value containing either a
* File instance or a String representing the original file path.
*
* @param deleteSourceFiles true to delete the source files.
*/
public void setDeleteSourceFiles(boolean deleteSourceFiles) {
@@ -303,7 +288,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
/**
* Set the charset name to use when writing a File from a String-based
* Message payload.
*
* @param charset The charset.
*/
public void setCharset(String charset) {
@@ -452,7 +436,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* 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
* @param newFileCallback a {@link BiConsumer} callback to be invoked when new file is created.
* @since 5.1
*/
public void setNewFileCallback(final BiConsumer<File, Message<?>> newFileCallback) {
@@ -464,8 +448,9 @@ 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));
final File directory =
new File(this.destinationDirectoryExpression.getValue(this.evaluationContext, null, String.class));
validateDestinationDirectory(directory, this.autoCreateDirectory);
}
@@ -638,8 +623,9 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
return null;
}
private File handleFileMessage(final File sourceFile, File tempFile, final File resultFile, final Message<?> requestMessage)
private File handleFileMessage(File sourceFile, File tempFile, File resultFile, Message<?> requestMessage)
throws IOException {
if (!FileExistsMode.APPEND.equals(this.fileExistsMode) && this.deleteSourceFiles) {
rename(sourceFile, resultFile);
return resultFile;
@@ -650,13 +636,15 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
}
}
private File handleInputStreamMessage(final InputStream sourceFileInputStream, File originalFile, File tempFile,
final File resultFile, final Message<?> requestMessage) throws IOException {
final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode)
|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
private File handleInputStreamMessage(InputStream sourceFileInputStream, File originalFile, File tempFile,
File resultFile, Message<?> requestMessage) throws IOException {
boolean append =
FileExistsMode.APPEND.equals(this.fileExistsMode)
|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
if (append) {
final File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile);
final File fileToWriteTo = determineFileToWrite(resultFile, tempFile);
WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry,
fileToWriteTo.getAbsolutePath()) {
@@ -677,7 +665,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
bos.write(buffer, 0, bytesRead);
}
if (FileWritingMessageHandler.this.appendNewLine) {
bos.write(LINE_SEPARATOR.getBytes());
bos.write(System.lineSeparator().getBytes());
}
}
finally {
@@ -718,7 +706,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
bos.write(buffer, 0, bytesRead);
}
if (this.appendNewLine) {
bos.write(LINE_SEPARATOR.getBytes());
bos.write(System.lineSeparator().getBytes());
}
bos.flush();
}
@@ -741,20 +729,20 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
}
}
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);
private File handleByteArrayMessage(byte[] bytes, File originalFile, File tempFile, File resultFile,
Message<?> requestMessage) throws IOException {
final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode);
final File fileToWriteTo = determineFileToWrite(resultFile, tempFile);
boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode)
|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry,
fileToWriteTo.getAbsolutePath()) {
@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))) {
if (append && FileWritingMessageHandler.this.newFileCallback != null && !fileToWriteTo.exists()) {
FileWritingMessageHandler.this.newFileCallback.accept(fileToWriteTo, requestMessage);
}
@@ -764,7 +752,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
bos = state != null ? state.stream : createOutputStream(fileToWriteTo, append);
bos.write(bytes);
if (FileWritingMessageHandler.this.appendNewLine) {
bos.write(LINE_SEPARATOR.getBytes());
bos.write(System.lineSeparator().getBytes());
}
}
finally {
@@ -790,20 +778,20 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
return resultFile;
}
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);
private File handleStringMessage(String content, File originalFile, File tempFile, File resultFile,
Message<?> requestMessage) throws IOException {
final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode);
File fileToWriteTo = determineFileToWrite(resultFile, tempFile);
boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode)
|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry,
fileToWriteTo.getAbsolutePath()) {
@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))) {
if (append && FileWritingMessageHandler.this.newFileCallback != null && !fileToWriteTo.exists()) {
FileWritingMessageHandler.this.newFileCallback.accept(fileToWriteTo, requestMessage);
}
@@ -914,8 +902,9 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
return destinationDirectory;
}
private synchronized FileState getFileState(final File fileToWriteTo, boolean isString)
private synchronized FileState getFileState(File fileToWriteTo, boolean isString)
throws FileNotFoundException {
FileState state;
boolean appendNoFlush = FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
if (appendNoFlush) {
@@ -953,7 +942,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* @throws FileNotFoundException if the file does not exist.
* @since 4.3.8
*/
protected BufferedWriter createWriter(final File fileToWriteTo, final boolean append) throws FileNotFoundException {
protected BufferedWriter createWriter(File fileToWriteTo, boolean append) throws FileNotFoundException {
return new BufferedWriter(new OutputStreamWriter(new FileOutputStream(fileToWriteTo, append), this.charset),
this.bufferSize);
}
@@ -968,6 +957,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
*/
protected BufferedOutputStream createOutputStream(File fileToWriteTo, final boolean append)
throws FileNotFoundException {
return new BufferedOutputStream(new FileOutputStream(fileToWriteTo, append), this.bufferSize);
}

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

@@ -56,7 +56,6 @@ import java.util.regex.Matcher;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.commons.logging.Log;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
@@ -117,15 +116,10 @@ public class FileWritingMessageHandlerTests {
private FileWritingMessageHandler handler;
@Before
public void setup() throws Exception {
//don't tamper with temp files here, Rule is applied later
}
@Test(expected = MessageHandlingException.class)
public void unsupportedType() throws Exception {
handler.handleMessage(new GenericMessage<Integer>(99));
assertThat(outputDirectory.listFiles()[0], nullValue());
public void unsupportedType() {
this.handler.handleMessage(new GenericMessage<>(99));
assertThat(this.outputDirectory.listFiles()[0], nullValue());
}
@Test

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

@@ -616,8 +616,8 @@ 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.
Starting with version 5.1, you can provide a `BiConsumer<File, Message<?>>` `newFileCallback` which is triggered if you use `FileExistsMode.APPEND` or `FileExistsMode.APPEND_NO_FLUSH` and a new file has to be created.
This callback receives a 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]]

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

@@ -13,7 +13,7 @@ The following components are new in 5.1:
* <<x5.1-AmqpDedicatedChannelAdvice>>
<<x5.1-AmqpDedicatedChannelAdvice>>
[[x5.1-AmqpDedicatedChannelAdvice]]
==== `AmqpDedicatedChannelAdvice`
See <<amqp-strict-ordering>>.
@@ -69,7 +69,7 @@ See <<json-transformers>> for more information.
Starting with version 5.0.5, generated bean names for the components in an `IntegrationFlow` include the flow bean name, followed by a dot, as a prefix. For example, if a flow bean were named `flowBean`, a generated bean might be named `flowBean.generatedBean`.
See "`<<java-dsl-flows>>`" for more information.
See <<java-dsl-flows>> for more information.
[[x5.1-aggregator]]
==== Aggregator Changes
@@ -77,22 +77,24 @@ See "`<<java-dsl-flows>>`" for more information.
If the `groupTimeout` is evaluated to a negative value, an aggregator now expires the group immediately.
Only `null` is considered as a signal to do nothing for the current message.
See "`<<aggregator>>`" for more information.
See <<aggregator>> for more information.
[[x5.1-publisher]]
==== @Publisher annotation changes
Starting with version 5.1, you must explicitly turn on the `@Publisher` AOP functionality by using `@EnablePublisher` or by using the `<int:enable-publisher>` child element on `<int:annotation-config>`.
See "`<<publisher-annotation>>`" for more information.
See <<publisher-annotation>> for more information.
[[x5.1-file-writing-message-handler]]
=== `FileWritingMessageHandler`: New file callback
[[x5.1-files]]
=== Files Changes
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.
See <<files>> for more information.
[[x5.1-amqp]]
=== AMQP Changes
@@ -107,7 +109,7 @@ See <<amqp-content-type>> for more information.
A confusing `max-rows-per-poll` property on the JDBC Inbound Channel Adapter and JDBC Outbound Gateway has been deprecated in favor of the newly introduced `max-rows` property.
See "`<<jdbc>>`" for more information.
See <<jdbc>> for more information.
[[x5.1-ftp-sftp]]
=== FTP and SFTP Changes
@@ -122,7 +124,8 @@ See <<ftp-streaming>> and <<sftp-streaming>> for more information.
In addition, the synchronizers for inbound channel adapters can now be provided with a `Comparator`.
This is useful when using `maxFetchSize` to limit the files retrieved.
==== Twitter Support
[[x5.1-twitter]]
=== Twitter Support
Since the Spring Social project has moved to https://spring.io/blog/2018/07/03/spring-social-end-of-life-announcement[end of life status], Twitter support in Spring Integration has been moved to the Extensions project.
See https://github.com/spring-projects/spring-integration-extensions/tree/master/spring-integration-social-twitter[Spring Integration Social Twitter] for more information.