From f71d6a0e66887eba75ad4dd634b3220c6aa74d0e Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 4 Nov 2016 16:23:43 -0400 Subject: [PATCH] INT-4154: Merge Files Java DSL JIRA: https://jira.spring.io/browse/INT-4154 * Move `Transformers` for files to the `Files` factory * Remove `@Deprecated` methods * Fix `FilesTests` for new packages and removed methods like `.handleWithAdapter()` Provide JavaDocs for File DSL and implement some new methods like `chmod()` and `renameFunction()` Fix race condition in the `ScriptsTests` when even small time window delay may lead us to one more message in the queue --- .../integration/dsl/Transformers.java | 19 + .../dsl/FileInboundChannelAdapterSpec.java | 229 +++++++++ .../file/dsl/FileSplitterSpec.java | 125 +++++ .../FileTransferringMessageHandlerSpec.java | 238 ++++++++++ .../dsl/FileWritingMessageHandlerSpec.java | 242 ++++++++++ .../integration/file/dsl/Files.java | 229 +++++++++ .../RemoteFileInboundChannelAdapterSpec.java | 226 +++++++++ .../dsl/RemoteFileOutboundGatewaySpec.java | 287 ++++++++++++ .../integration/file/dsl/TailAdapterSpec.java | 186 ++++++++ .../integration/file/dsl/package-info.java | 4 + .../AbstractRemoteFileOutboundGateway.java | 56 ++- .../FileTransferringMessageHandler.java | 13 + .../AbstractInboundFileSynchronizer.java | 4 - ...SDelegatingFileTailingMessageProducer.java | 6 +- .../integration/file/dsl/FileTests.java | 440 ++++++++++++++++++ .../scripting/dsl/ScriptsTests.java | 4 +- 16 files changed, 2286 insertions(+), 22 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileTransferringMessageHandlerSpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/Files.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileOutboundGatewaySpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/dsl/package-info.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/dsl/FileTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java index 69642a0a3d..77ef35afa7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java @@ -36,6 +36,7 @@ import org.springframework.integration.transformer.ObjectToStringTransformer; import org.springframework.integration.transformer.PayloadDeserializingTransformer; import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.integration.transformer.PayloadTypeConvertingTransformer; +import org.springframework.integration.transformer.StreamTransformer; import org.springframework.integration.transformer.SyslogToMapTransformer; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -231,4 +232,22 @@ public abstract class Transformers { return new DecodingTransformer<>(codec, typeExpression); } + /** + * The factory method for the {@link StreamTransformer}. + * @return the {@link StreamTransformer} instance. + */ + public static StreamTransformer fromStream() { + return fromStream(null); + } + + /** + * Create an instance with the charset to convert the stream to a + * String; if null a {@code byte[]} will be produced instead. + * @param charset the charset. + * @return the {@link StreamTransformer} instance. + */ + public static StreamTransformer fromStream(String charset) { + return new StreamTransformer(charset); + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..05a1c8fbf4 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java @@ -0,0 +1,229 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.io.File; +import java.util.Comparator; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.integration.dsl.MessageSourceSpec; +import org.springframework.integration.file.DirectoryScanner; +import org.springframework.integration.file.FileLocker; +import org.springframework.integration.file.FileReadingMessageSource; +import org.springframework.integration.file.config.FileListFilterFactoryBean; +import org.springframework.integration.file.filters.AcceptAllFileListFilter; +import org.springframework.integration.file.filters.AcceptOnceFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.IgnoreHiddenFileListFilter; +import org.springframework.integration.file.filters.RegexPatternFileListFilter; +import org.springframework.integration.file.filters.SimplePatternFileListFilter; +import org.springframework.integration.file.locking.NioFileLocker; +import org.springframework.util.Assert; + +/** + * A {@link MessageSourceSpec} for a {@link FileReadingMessageSource}. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class FileInboundChannelAdapterSpec + extends MessageSourceSpec { + + private final FileListFilterFactoryBean fileListFilterFactoryBean = new FileListFilterFactoryBean(); + + private FileLocker locker; + + FileInboundChannelAdapterSpec() { + this.target = new FileReadingMessageSource(); + } + + FileInboundChannelAdapterSpec(Comparator receptionOrderComparator) { + this.target = new FileReadingMessageSource(receptionOrderComparator) { + + @Override + protected void onInit() { + try { + setFilter(FileInboundChannelAdapterSpec.this.fileListFilterFactoryBean.getObject()); + } + catch (Exception e) { + throw new BeanCreationException("The bean for the [" + this + "] can not be instantiated.", e); + } + super.onInit(); + } + + }; + } + + /** + * Specify the input directory. + * @param directory the directory. + * @return the spec. + * @see FileReadingMessageSource#setDirectory(File) + */ + FileInboundChannelAdapterSpec directory(File directory) { + this.target.setDirectory(directory); + return _this(); + } + + /** + * Specify a custom scanner. + * @param scanner the scanner. + * @return the spec. + * @see FileReadingMessageSource#setScanner(DirectoryScanner) + */ + public FileInboundChannelAdapterSpec scanner(DirectoryScanner scanner) { + this.target.setScanner(scanner); + return _this(); + } + + /** + * Specify whether to create the source directory automatically if it does + * not yet exist upon initialization. By default, this value is + * true. If set to false and the + * source directory does not exist, an Exception will be thrown upon + * initialization. + * @param autoCreateDirectory the autoCreateDirectory. + * @return the spec. + * @see FileReadingMessageSource#setAutoCreateDirectory(boolean) + */ + public FileInboundChannelAdapterSpec autoCreateDirectory(boolean autoCreateDirectory) { + this.target.setAutoCreateDirectory(autoCreateDirectory); + return _this(); + } + + /** + * Configure the filter. + * @param filter the filter. + * @return the spec. + * @see FileReadingMessageSource#setFilter(FileListFilter) + */ + public FileInboundChannelAdapterSpec filter(FileListFilter filter) { + this.fileListFilterFactoryBean.setFilter(filter); + return _this(); + } + + /** + * Configure an {@link AcceptOnceFileListFilter} if {@code preventDuplicates == true}, + * otherwise - {@link AcceptAllFileListFilter}. + * @param preventDuplicates true to configure an {@link AcceptOnceFileListFilter}. + * @return the spec. + */ + public FileInboundChannelAdapterSpec preventDuplicates(boolean preventDuplicates) { + this.fileListFilterFactoryBean.setPreventDuplicates(preventDuplicates); + return _this(); + } + + /** + /** + * Configure an {@link IgnoreHiddenFileListFilter} if {@code ignoreHidden == true}. + * @param ignoreHidden true to configure an {@link IgnoreHiddenFileListFilter}. + * @return the spec. + */ + public FileInboundChannelAdapterSpec ignoreHidden(boolean ignoreHidden) { + this.fileListFilterFactoryBean.setIgnoreHidden(ignoreHidden); + return _this(); + } + + /** + * Configure a {@link SimplePatternFileListFilter}. + * @param pattern The pattern. + * @return the spec. + * @see FileReadingMessageSource#setFilter(FileListFilter) + * @see #filter(FileListFilter) + */ + public FileInboundChannelAdapterSpec patternFilter(String pattern) { + this.fileListFilterFactoryBean.setFilenamePattern(pattern); + return _this(); + } + + /** + * Configure a {@link RegexPatternFileListFilter}. + * @param regex The regex. + * @return the spec. + * @see FileReadingMessageSource#setFilter(FileListFilter) + * @see #filter(FileListFilter) + */ + public FileInboundChannelAdapterSpec regexFilter(String regex) { + this.fileListFilterFactoryBean.setFilenameRegex(regex); + return _this(); + } + + /** + * Set a {@link FileLocker} to be used to guard files against + * duplicate processing. + * @param locker the locker. + * @return the spec. + * @see FileReadingMessageSource#setLocker(FileLocker) + */ + public FileInboundChannelAdapterSpec locker(FileLocker locker) { + Assert.isNull(this.locker, + "The 'locker' (" + this.locker + ") is already configured for the FileReadingMessageSource"); + this.locker = locker; + this.target.setLocker(locker); + return _this(); + } + + /** + * Configure an {@link NioFileLocker}. + * @return the spec. + * @see #locker(FileLocker) + */ + public FileInboundChannelAdapterSpec nioLocker() { + return locker(new NioFileLocker()); + } + + /** + * Set this flag if you want to make sure the internal queue is + * refreshed with the latest content of the input directory on each poll. + * @param scanEachPoll the scanEachPoll. + * @return the spec. + * @see FileReadingMessageSource#setScanEachPoll(boolean) + */ + public FileInboundChannelAdapterSpec scanEachPoll(boolean scanEachPoll) { + this.target.setScanEachPoll(scanEachPoll); + return _this(); + } + + /** + * Switch this {@link FileReadingMessageSource} to use its internal + * {@link java.nio.file.WatchService} directory scanner. + * @param useWatchService the {@code boolean} flag to enable the use + * of a {@link java.nio.file.WatchService}. + * @return the spec. + * @see #watchEvents + * @see FileReadingMessageSource#setUseWatchService(boolean) + */ + public FileInboundChannelAdapterSpec useWatchService(boolean useWatchService) { + this.target.setUseWatchService(useWatchService); + return this; + } + + /** + * The {@link java.nio.file.WatchService} event types. + * If {@link #useWatchService} isn't {@code true}, this option is ignored. + * @param watchEvents the set of {@link FileReadingMessageSource.WatchEventType}. + * @return the spec. + * @see #useWatchService + * @see FileReadingMessageSource#setWatchEvents + */ + public FileInboundChannelAdapterSpec watchEvents(FileReadingMessageSource.WatchEventType... watchEvents) { + this.target.setWatchEvents(watchEvents); + return this; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java new file mode 100644 index 0000000000..4dbe1adf43 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileSplitterSpec.java @@ -0,0 +1,125 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.nio.charset.Charset; + +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.file.splitter.FileSplitter; + +/** + * The {@link MessageHandlerSpec} for the {@link FileSplitter}. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class FileSplitterSpec extends MessageHandlerSpec { + + private final boolean iterator; + + private boolean markers; + + private boolean markersJson; + + private Charset charset; + + private boolean applySequence; + + FileSplitterSpec() { + this(true); + } + + FileSplitterSpec(boolean iterator) { + this(iterator, false); + } + + FileSplitterSpec(boolean iterator, boolean markers) { + this.iterator = iterator; + this.markers = markers; + } + + /** + * Set the charset to be used when reading the file, when something other than the default + * charset is required. + * @param charset the charset. + * @return the FileSplitterSpec + */ + public FileSplitterSpec charset(String charset) { + return charset(Charset.forName(charset)); + } + + /** + * Set the charset to be used when reading the file, when something other than the default + * charset is required. + * @param charset the charset. + * @return the FileSplitterSpec + */ + public FileSplitterSpec charset(Charset charset) { + this.charset = charset; + return this; + } + + /** + * Specify if {@link FileSplitter} should emit + * {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker}s + * Defaults to {@code false}. + * @return the FileSplitterSpec + * @see FileSplitter + */ + public FileSplitterSpec markers() { + return markers(false); + } + + /** + * Specify if {@link FileSplitter} should emit + * {@link org.springframework.integration.file.splitter.FileSplitter.FileMarker}s + * and if they should be converted to the JSON string representation. + * Defaults to {@code false} for markers and {@code false} for markersJson. + * @param asJson the asJson flag to use. + * @return the FileSplitterSpec + * @see FileSplitter + */ + public FileSplitterSpec markers(boolean asJson) { + this.markers = true; + this.markersJson = asJson; + return this; + } + + /** + * A {@code boolean} flag to indicate if {@code sequenceDetails} should be + * applied for messages based on the lines from file. + * Defaults to {@code false}. + * @param applySequence the applySequence flag to use. + * @return the FileSplitterSpec + * @see org.springframework.integration.splitter.AbstractMessageSplitter#setApplySequence(boolean) + */ + public FileSplitterSpec applySequence(boolean applySequence) { + this.applySequence = applySequence; + return this; + } + + + @Override + protected FileSplitter doGet() { + FileSplitter fileSplitter = new FileSplitter(this.iterator, this.markers, this.markersJson); + fileSplitter.setApplySequence(this.applySequence); + fileSplitter.setCharset(this.charset); + return fileSplitter; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileTransferringMessageHandlerSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileTransferringMessageHandlerSpec.java new file mode 100644 index 0000000000..0934c6350b --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileTransferringMessageHandlerSpec.java @@ -0,0 +1,238 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.nio.charset.Charset; +import java.util.Collection; +import java.util.Collections; +import java.util.function.Function; + +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.file.DefaultFileNameGenerator; +import org.springframework.integration.file.FileNameGenerator; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * The {@link MessageHandlerSpec} for the {@link FileTransferringMessageHandler}. + * + * @param the target file type. + * @param the target {@link FileTransferringMessageHandlerSpec} implementation type. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public abstract class FileTransferringMessageHandlerSpec> + extends MessageHandlerSpec> + implements ComponentsRegistration { + + private FileNameGenerator fileNameGenerator; + + private DefaultFileNameGenerator defaultFileNameGenerator; + + protected FileTransferringMessageHandlerSpec(SessionFactory sessionFactory) { + this.target = new FileTransferringMessageHandler<>(sessionFactory); + } + + protected FileTransferringMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate) { + this.target = new FileTransferringMessageHandler<>(remoteFileTemplate); + } + + protected FileTransferringMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate, + FileExistsMode fileExistsMode) { + this.target = new FileTransferringMessageHandler<>(remoteFileTemplate, fileExistsMode); + } + + /** + * A {@code boolean} flag to indicate automatically create the directory or not. + * @param autoCreateDirectory true to automatically create the directory. + * @return the current Spec + */ + public S autoCreateDirectory(boolean autoCreateDirectory) { + this.target.setAutoCreateDirectory(autoCreateDirectory); + return _this(); + } + + /** + * Specify a remote file separator symbol. + * @param remoteFileSeparator the remote file separator. + * @return the current Spec + */ + public S remoteFileSeparator(String remoteFileSeparator) { + this.target.setRemoteFileSeparator(remoteFileSeparator); + return _this(); + } + + /** + * Specify a remote directory path. + * @param remoteDirectory the remote directory path. + * @return the current Spec + */ + public S remoteDirectory(String remoteDirectory) { + this.target.setRemoteDirectoryExpression(new LiteralExpression(remoteDirectory)); + return _this(); + } + + /** + * Specify a remote directory path SpEL expression. + * @param remoteDirectoryExpression the remote directory expression + * @return the current Spec + */ + public S remoteDirectoryExpression(String remoteDirectoryExpression) { + this.target.setRemoteDirectoryExpression(PARSER.parseExpression(remoteDirectoryExpression)); + return _this(); + } + + /** + * Specify a remote directory path {@link Function}. + * @param remoteDirectoryFunction the remote directory {@link Function} + * @param

the expected payload type. + * @return the current Spec + */ + public

S remoteDirectory(Function, String> remoteDirectoryFunction) { + this.target.setRemoteDirectoryExpression(new FunctionExpression<>(remoteDirectoryFunction)); + return _this(); + } + + /** + * Specify a remote directory path. + * @param temporaryRemoteDirectory the temporary remote directory path + * @return the current Spec + */ + public S temporaryRemoteDirectory(String temporaryRemoteDirectory) { + this.target.setTemporaryRemoteDirectoryExpression(new LiteralExpression(temporaryRemoteDirectory)); + return _this(); + } + + /** + * Specify a remote directory path SpEL expression. + * @param temporaryRemoteDirectoryExpression the temporary remote directory path SpEL expression + * @return the current Spec + */ + public S temporaryRemoteDirectoryExpression(String temporaryRemoteDirectoryExpression) { + this.target.setTemporaryRemoteDirectoryExpression(PARSER.parseExpression(temporaryRemoteDirectoryExpression)); + return _this(); + } + + /** + * Specify a remote temporary directory path {@link Function}. + * @param temporaryRemoteDirectoryFunction the temporary remote directory {@link Function} + * @param

the expected payload type. + * @return the current Spec + */ + public

S temporaryRemoteDirectory(Function, String> temporaryRemoteDirectoryFunction) { + this.target.setTemporaryRemoteDirectoryExpression(new FunctionExpression<>(temporaryRemoteDirectoryFunction)); + return _this(); + } + + /** + * A {@code boolean} flag to use temporary files names or not. + * Defaults to {@code true}. + * @param useTemporaryFileName true to use a temporary file name. + * @return the current Spec + */ + public S useTemporaryFileName(boolean useTemporaryFileName) { + this.target.setUseTemporaryFileName(useTemporaryFileName); + return _this(); + } + + /** + * Set the file name generator used to generate the remote filename to be used when transferring + * files to the remote system. Default {@link DefaultFileNameGenerator}. + * @param fileNameGenerator the file name generator. + * @return the current Spec + */ + public S fileNameGenerator(FileNameGenerator fileNameGenerator) { + this.fileNameGenerator = fileNameGenerator; + this.target.setFileNameGenerator(fileNameGenerator); + return _this(); + } + + /** + * Set the {@link DefaultFileNameGenerator} based on the provided SpEL expression. + * @param fileNameGeneratorExpression the SpEL expression for file names generation. + * @return the current Spec + */ + public S fileNameExpression(String fileNameGeneratorExpression) { + Assert.isNull(this.fileNameGenerator, + "'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive."); + this.defaultFileNameGenerator = new DefaultFileNameGenerator(); + this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression); + return fileNameGenerator(this.defaultFileNameGenerator); + } + + /** + * Set the charset to use when converting String payloads to bytes as the content of the + * remote file. Default {@code UTF-8}. + * @param charset the charset. + * @return the current Spec + */ + public S charset(String charset) { + this.target.setCharset(charset); + return _this(); + } + + /** + * Set the charset to use when converting String payloads to bytes as the content of the + * remote file. Default {@code UTF-8}. + * @param charset the charset. + * @return the current Spec + */ + public S charset(Charset charset) { + Assert.notNull(charset, "'charset' must not be null."); + return charset(charset.name()); + } + + /** + * Set the temporary suffix to use when transferring files to the remote system. + * Default ".writing". + * @param temporaryFileSuffix the suffix + * @return the current Spec + */ + public S temporaryFileSuffix(String temporaryFileSuffix) { + this.target.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + /** + * Set the file permissions after uploading, e.g. 0600 for + * owner read/write. + * @param chmod the permissions. + * @return the current Spec + */ + public S chmod(int chmod) { + this.target.setChmod(chmod); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + if (this.defaultFileNameGenerator != null) { + return Collections.singletonList(this.defaultFileNameGenerator); + } + return null; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java new file mode 100644 index 0000000000..bc13124194 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileWritingMessageHandlerSpec.java @@ -0,0 +1,242 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.io.File; +import java.util.Collection; +import java.util.Collections; +import java.util.function.Function; + +import org.springframework.expression.Expression; +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.file.DefaultFileNameGenerator; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.FileNameGenerator; +import org.springframework.integration.file.FileWritingMessageHandler; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.messaging.Message; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; + +/** + * The {@link MessageHandlerSpec} for the {@link FileWritingMessageHandler}. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class FileWritingMessageHandlerSpec + extends MessageHandlerSpec + implements ComponentsRegistration { + + private FileNameGenerator fileNameGenerator; + + private DefaultFileNameGenerator defaultFileNameGenerator; + + FileWritingMessageHandlerSpec(File destinationDirectory) { + this.target = new FileWritingMessageHandler(destinationDirectory); + } + + FileWritingMessageHandlerSpec(String directoryExpression) { + this(PARSER.parseExpression(directoryExpression)); + } + +

FileWritingMessageHandlerSpec(Function, ?> directoryFunction) { + this(new FunctionExpression<>(directoryFunction)); + } + + FileWritingMessageHandlerSpec(Expression directoryExpression) { + this.target = new FileWritingMessageHandler(directoryExpression); + } + + FileWritingMessageHandlerSpec expectReply(boolean expectReply) { + this.target.setExpectReply(expectReply); + if (expectReply) { + this.target.setRequiresReply(true); + } + return _this(); + } + + /** + * Specify whether to create the destination directory automatically if it + * does not yet exist upon initialization. By default, this value is + * true. If set to false and the + * destination directory does not exist, an Exception will be thrown upon + * initialization. + * @param autoCreateDirectory true to create the directory if needed. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec autoCreateDirectory(boolean autoCreateDirectory) { + this.target.setAutoCreateDirectory(autoCreateDirectory); + return _this(); + } + + /** + * 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 {@code .writing}. + * @param temporaryFileSuffix The temporary file suffix. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec temporaryFileSuffix(String temporaryFileSuffix) { + this.target.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + /** + * Set the {@link FileExistsMode} that specifies what will happen in + * case the destination exists. + * @param fileExistsMode the {@link FileExistsMode} to consult. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec fileExistsMode(FileExistsMode fileExistsMode) { + this.target.setFileExistsMode(fileExistsMode); + return _this(); + } + + /** + * Set the file name generator used to generate the target file name. + * Default {@link DefaultFileNameGenerator}. + * @param fileNameGenerator the file name generator. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec fileNameGenerator(FileNameGenerator fileNameGenerator) { + this.fileNameGenerator = fileNameGenerator; + this.target.setFileNameGenerator(fileNameGenerator); + return _this(); + } + + /** + * Set the {@link DefaultFileNameGenerator} based on the provided SpEL expression. + * @param fileNameExpression the SpEL expression for file names generation. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec fileNameExpression(String fileNameExpression) { + Assert.isNull(this.fileNameGenerator, + "'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive."); + this.defaultFileNameGenerator = new DefaultFileNameGenerator(); + this.defaultFileNameGenerator.setExpression(fileNameExpression); + return fileNameGenerator(this.defaultFileNameGenerator); + } + + /** + * Specify whether to delete source Files after writing to the destination + * directory. The default is false. When set to true, it + * 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. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec deleteSourceFiles(boolean deleteSourceFiles) { + this.target.setDeleteSourceFiles(deleteSourceFiles); + return _this(); + } + + /** + * Set the charset to use when converting String payloads to bytes as the content of the file. + * Default {@code UTF-8}. + * @param charset the charset. + * @return the current Spec + */ + public FileWritingMessageHandlerSpec charset(String charset) { + this.target.setCharset(charset); + return _this(); + } + + /** + * If {@code true} will append a new-line after each write. + * Defaults to {@code false}. + * @param appendNewLine true if a new-line should be written to the file after payload is written. + * @return the spec. + * @see FileWritingMessageHandler#setAppendNewLine(boolean) + */ + public FileWritingMessageHandlerSpec appendNewLine(boolean appendNewLine) { + this.target.setAppendNewLine(appendNewLine); + return this; + } + + /** + * Set the buffer size to use while writing to files; default 8192. + * @param bufferSize the buffer size. + * @return the spec. + * @see FileWritingMessageHandler#setBufferSize(int) + */ + public FileWritingMessageHandlerSpec bufferSize(int bufferSize) { + this.target.setBufferSize(bufferSize); + return this; + } + + /** + * Set the frequency to flush buffers when {@link FileExistsMode#APPEND_NO_FLUSH} is + * being used. + * @param flushInterval the interval. + * @return the spec. + * @see FileWritingMessageHandler#setBufferSize(int) + */ + public FileWritingMessageHandlerSpec flushInterval(long flushInterval) { + this.target.setFlushInterval(flushInterval); + return this; + } + + /** + * Specify a {@link TaskScheduler} for flush task when the {@link FileExistsMode#APPEND_NO_FLUSH} is in use. + * @param taskScheduler the {@link TaskScheduler} to use. + * @return the spec. + * @see FileWritingMessageHandler#setTaskScheduler(TaskScheduler) + */ + public FileWritingMessageHandlerSpec taskScheduler(TaskScheduler taskScheduler) { + this.target.setTaskScheduler(taskScheduler); + return this; + } + + /** + * Specify a {@link FileWritingMessageHandler.MessageFlushPredicate} for flush task + * when the {@link FileExistsMode#APPEND_NO_FLUSH} is in use. + * @param flushPredicate the {@link FileWritingMessageHandler.MessageFlushPredicate} to use. + * @return the spec. + * @see FileWritingMessageHandler#setFlushPredicate(FileWritingMessageHandler.MessageFlushPredicate) + */ + public FileWritingMessageHandlerSpec flushPredicate( + FileWritingMessageHandler.MessageFlushPredicate flushPredicate) { + this.target.setFlushPredicate(flushPredicate); + return this; + } + + /** + * Set to true to preserve the destination file timestamp. If true and + * the payload is a {@link File}, the payload's {@code lastModified} time will be + * transferred to the destination file. + * @param preserveTimestamp the {@code boolean} flag to use. + * @return the spec. + * @see FileWritingMessageHandler#setPreserveTimestamp(boolean) + */ + public FileWritingMessageHandlerSpec preserveTimestamp(boolean preserveTimestamp) { + this.target.setPreserveTimestamp(preserveTimestamp); + return this; + } + + @Override + public Collection getComponentsToRegister() { + if (this.defaultFileNameGenerator != null) { + return Collections.singletonList(this.defaultFileNameGenerator); + } + return null; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/Files.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/Files.java new file mode 100644 index 0000000000..5942c96f1b --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/Files.java @@ -0,0 +1,229 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.io.File; +import java.util.Comparator; +import java.util.function.Function; + +import org.springframework.expression.Expression; +import org.springframework.integration.file.transformer.FileToByteArrayTransformer; +import org.springframework.integration.file.transformer.FileToStringTransformer; +import org.springframework.messaging.Message; + +/** + * The Spring Integration File components Factory. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public abstract class Files { + + /** + * Create a {@link FileInboundChannelAdapterSpec} builder for the {@code FileReadingMessageSource}. + * @param directory the directory to scan files. + * @return the {@link FileInboundChannelAdapterSpec} instance. + */ + public static FileInboundChannelAdapterSpec inboundAdapter(File directory) { + return inboundAdapter(directory, null); + } + + /** + * Create a {@link FileInboundChannelAdapterSpec} builder for the {@code FileReadingMessageSource}. + * @param directory the directory to scan files. + * @param receptionOrderComparator the {@link Comparator} for ordering file objects. + * @return the {@link FileInboundChannelAdapterSpec} instance. + */ + public static FileInboundChannelAdapterSpec inboundAdapter(File directory, + Comparator receptionOrderComparator) { + return new FileInboundChannelAdapterSpec(receptionOrderComparator).directory(directory); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the one-way {@code FileWritingMessageHandler}. + * @param destinationDirectory the target directory to write files. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static FileWritingMessageHandlerSpec outboundAdapter(File destinationDirectory) { + return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(false); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the one-way {@code FileWritingMessageHandler}. + * @param directoryExpression the SpEL expression to evaluate target directory for writing files. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static FileWritingMessageHandlerSpec outboundAdapter(String directoryExpression) { + return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(false); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the one-way {@code FileWritingMessageHandler}. + * @param directoryExpression an expression to evaluate the target directory. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static FileWritingMessageHandlerSpec outboundAdapter(Expression directoryExpression) { + return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(false); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the one-way {@code FileWritingMessageHandler}. + * @param directoryFunction the {@link Function} to evaluate the target directory at runtime. + * @param

the payload type. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static

FileWritingMessageHandlerSpec outboundAdapter(Function, ?> directoryFunction) { + return new FileWritingMessageHandlerSpec(directoryFunction).expectReply(false); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the gateway {@code FileWritingMessageHandler}. + * @param destinationDirectory the target directory to write files. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static FileWritingMessageHandlerSpec outboundGateway(File destinationDirectory) { + return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(true); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the gateway {@code FileWritingMessageHandler}. + * @param directoryExpression the SpEL expression to evaluate target directory for writing files. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static FileWritingMessageHandlerSpec outboundGateway(String directoryExpression) { + return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(true); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} based on the provided {@link Expression} for directory. + * @param directoryExpression an expression to evaluate the target directory. + * @return the FileWritingMessageHandlerSpec instance. + */ + public static FileWritingMessageHandlerSpec outboundGateway(Expression directoryExpression) { + return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(true); + } + + /** + * Create a {@link FileWritingMessageHandlerSpec} builder for the gateway {@code FileWritingMessageHandler}. + * @param directoryFunction the {@link Function} to evaluate the target directory at runtime. + * @param

the payload type. + * @return the {@link FileWritingMessageHandlerSpec} instance. + */ + public static

FileWritingMessageHandlerSpec outboundGateway(Function, ?> directoryFunction) { + return new FileWritingMessageHandlerSpec(directoryFunction).expectReply(true); + } + + /** + * Create a {@link TailAdapterSpec} builder for the {@code FileTailingMessageProducerSupport}. + * @param file the file to tail. + * @return the {@link TailAdapterSpec} instance. + */ + public static TailAdapterSpec tailAdapter(File file) { + return new TailAdapterSpec().file(file); + } + + /** + * The {@link FileSplitterSpec} builder factory method with default arguments. + * @return the {@link FileSplitterSpec} builder. + */ + public static FileSplitterSpec splitter() { + return splitter(true); + } + + /** + * The {@link FileSplitterSpec} builder factory method with {@code iterator} flag specified. + * @param iterator the {@code boolean} flag to specify the {@code iterator} mode or not. + * @return the {@link FileSplitterSpec} builder. + */ + public static FileSplitterSpec splitter(boolean iterator) { + return splitter(iterator, false); + } + + /** + * The {@link FileSplitterSpec} builder factory method with {@code iterator} and {@code markers} + * flags specified. + * @param iterator the {@code boolean} flag to specify the {@code iterator} mode or not. + * @param markers true to emit start of file/end of file marker messages before/after the data. + * @return the {@link FileSplitterSpec} builder. + */ + public static FileSplitterSpec splitter(boolean iterator, boolean markers) { + return new FileSplitterSpec(iterator, markers); + } + + /** + * Create a {@link FileToStringTransformer} instance with default {@code charset} and no delete files afterwards. + * @return the {@link FileToStringTransformer}. + */ + public static FileToStringTransformer toStringTransformer() { + return toStringTransformer(false); + } + + /** + * Create a {@link FileToStringTransformer} instance with default {@code charset} and with delete files flag. + * @param deleteFiles true to delete the file. + * @return the {@link FileToStringTransformer}. + */ + public static FileToStringTransformer toStringTransformer(boolean deleteFiles) { + return toStringTransformer(null, deleteFiles); + } + + /** + * Create a {@link FileToStringTransformer} instance with provided {@code charset} and no delete files afterwards. + * @param charset The charset. + * @return the {@link FileToStringTransformer}. + */ + public static FileToStringTransformer toStringTransformer(String charset) { + return toStringTransformer(charset, false); + } + + /** + * Create a {@link FileToStringTransformer} instance with provided {@code charset} and delete files flag. + * @param charset The charset. + * @param deleteFiles true to delete the file. + * @return the {@link FileToStringTransformer}. + */ + public static FileToStringTransformer toStringTransformer(String charset, boolean deleteFiles) { + FileToStringTransformer transformer = new FileToStringTransformer(); + if (charset != null) { + transformer.setCharset(charset); + } + transformer.setDeleteFiles(deleteFiles); + return transformer; + } + + /** + * Create a {@link FileToByteArrayTransformer} instance. + * @return the {@link FileToByteArrayTransformer}. + */ + public static FileToByteArrayTransformer toByteArrayTransformer() { + return toByteArrayTransformer(false); + } + + /** + * Create a {@link FileToByteArrayTransformer} instance. + * @param deleteFiles specify whether to delete the File after transformation. + * Default is false. + * @return the {@link FileToByteArrayTransformer}. + */ + public static FileToByteArrayTransformer toByteArrayTransformer(boolean deleteFiles) { + FileToByteArrayTransformer fileToByteArrayTransformer = new FileToByteArrayTransformer(); + fileToByteArrayTransformer.setDeleteFiles(deleteFiles); + return fileToByteArrayTransformer; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java new file mode 100644 index 0000000000..b927e49b50 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java @@ -0,0 +1,226 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.io.File; +import java.util.Collection; +import java.util.Collections; +import java.util.function.Function; + +import org.springframework.expression.Expression; +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageSourceSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.file.filters.CompositeFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer; +import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource; + +/** + * A {@link MessageSourceSpec} for an {@link AbstractInboundFileSynchronizingMessageSource}. + * + * @param the target file type. + * @param the target {@link RemoteFileInboundChannelAdapterSpec} implementation type. + * @param the target {@link AbstractInboundFileSynchronizingMessageSource} implementation type. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public abstract class RemoteFileInboundChannelAdapterSpec, + MS extends AbstractInboundFileSynchronizingMessageSource> + extends MessageSourceSpec implements ComponentsRegistration { + + protected final AbstractInboundFileSynchronizer synchronizer; + + private CompositeFileListFilter filter; + + protected RemoteFileInboundChannelAdapterSpec(AbstractInboundFileSynchronizer synchronizer) { + this.synchronizer = synchronizer; + } + + /** + * Configure whether the local directory should be created by the adapter. + * @param autoCreateLocalDirectory the autoCreateLocalDirectory + * @return the spec. + */ + public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) { + this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory); + return _this(); + } + + /** + * Configure the local directory to copy files to. + * @param localDirectory the localDirectory. + * @return the spec. + */ + public S localDirectory(File localDirectory) { + this.target.setLocalDirectory(localDirectory); + return _this(); + } + + /** + * A {@link FileListFilter} used to determine which files will generate messages + * after they have been synchronized. + * @param localFileListFilter the localFileListFilter. + * @return the spec. + * @see AbstractInboundFileSynchronizingMessageSource#setLocalFilter(FileListFilter) + */ + public S localFilter(FileListFilter localFileListFilter) { + this.target.setLocalFilter(localFileListFilter); + return _this(); + } + + /** + * Configure the file name path separator used by the remote system. Defaults to '/'. + * @param remoteFileSeparator the remoteFileSeparator. + * @return the spec. + */ + public S remoteFileSeparator(String remoteFileSeparator) { + this.synchronizer.setRemoteFileSeparator(remoteFileSeparator); + return _this(); + } + + /** + * Configure a SpEL expression to generate the local file name; the root object for + * the evaluation is the remote file name. + * @param localFilenameExpression the localFilenameExpression. + * @return the spec. + */ + public S localFilenameExpression(String localFilenameExpression) { + return localFilenameExpression(PARSER.parseExpression(localFilenameExpression)); + } + + /** + * Configure a {@link Function} to be invoked to generate the local file name; + * argument passed to the {@code apply} method is the remote file name. + * @param localFilenameFunction the localFilenameFunction. + * @return the spec. + * @see FunctionExpression + */ + public S localFilename(Function localFilenameFunction) { + return localFilenameExpression(new FunctionExpression<>(localFilenameFunction)); + } + + /** + * Configure a SpEL expression to generate the local file name; the root object for + * the evaluation is the remote file name. + * @param localFilenameExpression the localFilenameExpression. + * @return the spec. + */ + public S localFilenameExpression(Expression localFilenameExpression) { + this.synchronizer.setLocalFilenameGeneratorExpression(localFilenameExpression); + return _this(); + } + + /** + * Configure a suffix to temporarily apply to the local filename; when copied the + * file is renamed to its final name. Default: '.writing'. + * @param temporaryFileSuffix the temporaryFileSuffix. + * @return the spec. + */ + public S temporaryFileSuffix(String temporaryFileSuffix) { + this.synchronizer.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + /** + * Specify the full path to the remote directory. + * @param remoteDirectory the remoteDirectory. + * @return the spec. + * @see AbstractInboundFileSynchronizer#setRemoteDirectory(String) + */ + public S remoteDirectory(String remoteDirectory) { + this.synchronizer.setRemoteDirectory(remoteDirectory); + return _this(); + } + + /** + * Specify an expression that evaluates to the full path to the remote directory. + * @param remoteDirectoryExpression The remote directory expression. + * @return the spec. + */ + public S remoteDirectoryExpression(Expression remoteDirectoryExpression) { + this.synchronizer.setRemoteDirectoryExpression(remoteDirectoryExpression); + return _this(); + } + + /** + * Configure a {@link FileListFilter} to be applied to the remote files before + * copying them. + * @param filter the filter. + * @return the spec. + */ + public S filter(FileListFilter filter) { + if (this.filter == null) { + if (filter instanceof CompositeFileListFilter) { + this.filter = (CompositeFileListFilter) filter; + } + else { + this.filter = new CompositeFileListFilter(); + this.filter.addFilter(filter); + } + this.synchronizer.setFilter(this.filter); + } + else { + this.filter.addFilter(filter); + } + return _this(); + } + + /** + * Configure a simple pattern filter (e.g. '*.txt'). + * @param pattern the pattern. + * @return the spec. + * @see #filter(FileListFilter) + */ + public abstract S patternFilter(String pattern); + + /** + * Configure a regex pattern filter (e.g. '[0-9].*.txt'). + * @param regex the regex. + * @return the spec. + * @see #filter(FileListFilter) + */ + public abstract S regexFilter(String regex); + + /** + * Set to true to enable deletion of remote files after successful transfer. + * @param deleteRemoteFiles true to delete. + * @return the spec. + */ + public S deleteRemoteFiles(boolean deleteRemoteFiles) { + this.synchronizer.setDeleteRemoteFiles(deleteRemoteFiles); + return _this(); + } + + /** + * Set to true to enable the preservation of the remote file timestamp when transferring. + * @param preserveTimestamp true to preserve. + * @return the spec. + */ + public S preserveTimestamp(boolean preserveTimestamp) { + this.synchronizer.setPreserveTimestamp(preserveTimestamp); + return _this(); + } + + @Override + public Collection getComponentsToRegister() { + return Collections.singletonList(this.synchronizer); + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileOutboundGatewaySpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileOutboundGatewaySpec.java new file mode 100644 index 0000000000..c4b5fe88e1 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileOutboundGatewaySpec.java @@ -0,0 +1,287 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.io.File; +import java.util.function.Function; + +import org.springframework.expression.Expression; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.file.filters.CompositeFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.RegexPatternFileListFilter; +import org.springframework.integration.file.filters.SimplePatternFileListFilter; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.messaging.Message; + +/** + * The {@link MessageHandlerSpec} for the {@link AbstractRemoteFileOutboundGateway}. + * + * @param the target file type. + * @param the target {@link RemoteFileOutboundGatewaySpec} implementation type. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public abstract class RemoteFileOutboundGatewaySpec> + extends MessageHandlerSpec> { + + private CompositeFileListFilter filter; + + private CompositeFileListFilter mputFilter; + + protected RemoteFileOutboundGatewaySpec(AbstractRemoteFileOutboundGateway outboundGateway) { + this.target = outboundGateway; + this.target.setRequiresReply(true); + } + + /** + * Specify the array of options for various gateway commands. + * @param options the options to set. + * @return the spec + * @see #options(AbstractRemoteFileOutboundGateway.Option...) + */ + public S options(String options) { + this.target.setOptions(options); + return _this(); + } + + /** + * Specify the array of {@link AbstractRemoteFileOutboundGateway.Option} + * for various gateway commands. + * @param options the options to set. + * @return the spec + */ + public S options(AbstractRemoteFileOutboundGateway.Option... options) { + this.target.setOptions(options); + return _this(); + } + + /** + * Set the file separator when dealing with remote files; default '/'. + * @param remoteFileSeparator the separator. + * @return the spec + */ + public S remoteFileSeparator(String remoteFileSeparator) { + this.target.setRemoteFileSeparator(remoteFileSeparator); + return _this(); + } + + /** + * Specify a directory path where remote files will be transferred to. + * @param localDirectory the localDirectory to set + * @return the spec + */ + public S localDirectory(File localDirectory) { + this.target.setLocalDirectory(localDirectory); + return _this(); + } + + /** + * Specify a SpEL expression to evaluate directory path where remote files will be transferred to. + * @param localDirectoryExpression the SpEL to determine the local directory. + * @return the spec + */ + public S localDirectoryExpression(String localDirectoryExpression) { + return localDirectoryExpression(PARSER.parseExpression(localDirectoryExpression)); + } + + /** + * Specify a {@link Function} to evaluate directory path where remote files will be transferred to. + * @param localDirectoryFunction the {@link Function} to determine the local directory. + * @param

the expected payload type. + * @return the spec + */ + public

S localDirectory(Function, String> localDirectoryFunction) { + return localDirectoryExpression(new FunctionExpression<>(localDirectoryFunction)); + } + + /** + * Specify a SpEL expression to evaluate directory path where remote files will be transferred to. + * @param localDirectoryExpression a SpEL expression to evaluate the local directory. + * @return the Spec. + */ + public S localDirectoryExpression(Expression localDirectoryExpression) { + this.target.setLocalDirectoryExpression(localDirectoryExpression); + return _this(); + } + + /** + * A {@code boolean} flag to identify if local directory should be created automatically. + * Defaults to {@code true}. + * @param autoCreateLocalDirectory the autoCreateLocalDirectory to set + * @return the Spec. + */ + public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) { + this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory); + return _this(); + } + + /** + * Set the temporary suffix to use when transferring files to the remote system. + * Default {@code .writing}. + * @param temporaryFileSuffix the temporaryFileSuffix to set + * @return the Spec. + */ + public S temporaryFileSuffix(String temporaryFileSuffix) { + this.target.setTemporaryFileSuffix(temporaryFileSuffix); + return _this(); + } + + /** + * Set a {@link FileListFilter} to filter remote files. + * @param filter the filter to set + * @return the Spec. + */ + public S filter(FileListFilter filter) { + if (this.filter == null) { + if (filter instanceof CompositeFileListFilter) { + this.filter = (CompositeFileListFilter) filter; + } + else { + this.filter = new CompositeFileListFilter<>(); + this.filter.addFilter(filter); + } + this.target.setFilter(this.filter); + } + else { + this.filter.addFilter(filter); + } + return _this(); + } + + /** + * A {@link FileListFilter} that runs against the local file system view when + * using {@code MPUT} command. + * @param filter the filter to set + * @return the Spec. + */ + public S mputFilter(FileListFilter filter) { + if (this.mputFilter == null) { + if (filter instanceof CompositeFileListFilter) { + this.mputFilter = (CompositeFileListFilter) filter; + } + else { + this.mputFilter = new CompositeFileListFilter<>(); + this.mputFilter.addFilter(filter); + } + this.target.setMputFilter(this.mputFilter); + } + else { + this.mputFilter.addFilter(filter); + } + return _this(); + } + + /** + * A {@link SimplePatternFileListFilter} that runs against the local file system view when + * using {@code MPUT} command. + * @param pattern the {@link SimplePatternFileListFilter} for {@code MPUT} command. + * @return the Spec. + */ + public S patternMputFilter(String pattern) { + return mputFilter(new SimplePatternFileListFilter(pattern)); + } + + /** + * A {@link SimplePatternFileListFilter} that runs against the local file system view when + * using {@code MPUT} command. + * @param regex the {@link SimplePatternFileListFilter} for {@code MPUT} command. + * @return the Spec. + */ + public S regexMpuFilter(String regex) { + return mputFilter(new RegexPatternFileListFilter(regex)); + } + + /** + * Specify a SpEL expression for files renaming during transfer. + * @param expression the String in SpEL syntax. + * @return the Spec. + */ + public S renameExpression(String expression) { + this.target.setRenameExpressionString(expression); + return _this(); + } + + /** + * Specify a SpEL expression for files renaming during transfer. + * @param expression the String in SpEL syntax. + * @return the Spec. + */ + public S renameExpression(Expression expression) { + this.target.setRenameExpression(expression); + return _this(); + } + + /** + * Specify a {@link Function} for files renaming during transfer. + * @param renameFunction the {@link Function} to use. + * @param

the expected payload type. + * @return the Spec. + */ + public

S renameFunction(Function, String> renameFunction) { + this.target.setRenameExpression(new FunctionExpression<>(renameFunction)); + return _this(); + } + + /** + * Specify a SpEL expression for local files renaming after downloading. + * @param localFilenameExpression the SpEL expression to use. + * @return the Spec. + */ + public S localFilenameExpression(String localFilenameExpression) { + return localFilenameExpression(PARSER.parseExpression(localFilenameExpression)); + } + + /** + * Specify a {@link Function} for local files renaming after downloading. + * @param localFilenameFunction the {@link Function} to use. + * @param

the expected payload type. + * @return the Spec. + */ + public

S localFilename(Function, String> localFilenameFunction) { + return localFilenameExpression(new FunctionExpression<>(localFilenameFunction)); + } + + /** + * Specify a SpEL expression for local files renaming after downloading. + * @param localFilenameExpression a SpEL expression to evaluate the local file name. + * @return the Spec. + */ + public S localFilenameExpression(Expression localFilenameExpression) { + this.target.setLocalFilenameGeneratorExpression(localFilenameExpression); + return _this(); + } + + /** + * Set the file permissions after uploading, e.g. 0600 for + * owner read/write. + * @param chmod the permissions. + * @return the current Spec + */ + public S chmod(int chmod) { + this.target.setChmod(chmod); + return _this(); + } + + public abstract S patternFileNameFilter(String pattern); + + public abstract S regexFileNameFilter(String regex); + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java new file mode 100644 index 0000000000..bcc893e11d --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java @@ -0,0 +1,186 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.dsl; + +import java.io.File; + +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.dsl.MessageProducerSpec; +import org.springframework.integration.file.config.FileTailInboundChannelAdapterFactoryBean; +import org.springframework.integration.file.tail.FileTailingMessageProducerSupport; +import org.springframework.messaging.MessageChannel; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; + +/** + * A {@link MessageProducerSpec} for file tailing adapters. + * + * @author Artem Bilan + * + * @since 5.0 + */ +public class TailAdapterSpec extends MessageProducerSpec { + + private final FileTailInboundChannelAdapterFactoryBean factoryBean = new FileTailInboundChannelAdapterFactoryBean(); + + private MessageChannel outputChannel; + + private MessageChannel errorChannel; + + TailAdapterSpec() { + super(null); + this.factoryBean.setBeanFactory(new DefaultListableBeanFactory()); + } + + TailAdapterSpec file(File file) { + Assert.notNull(file); + this.factoryBean.setFile(file); + return _this(); + } + + /** + * Specify the options string for native {@code tail} command. + * @param nativeOptions the nativeOptions. + * @return the spec. + * @see org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer#setOptions(String) + */ + public TailAdapterSpec nativeOptions(String nativeOptions) { + this.factoryBean.setNativeOptions(nativeOptions); + return _this(); + } + + /** + * Configure a task executor. Defaults to a + * {@link org.springframework.core.task.SimpleAsyncTaskExecutor}. + * @param taskExecutor the taskExecutor. + * @return the spec. + */ + public TailAdapterSpec taskExecutor(TaskExecutor taskExecutor) { + this.factoryBean.setTaskExecutor(taskExecutor); + return _this(); + } + + /** + * Set a task scheduler - defaults to the integration 'taskScheduler'. + * @param taskScheduler the taskScheduler. + * @return the spec. + */ + public TailAdapterSpec taskScheduler(TaskScheduler taskScheduler) { + this.factoryBean.setTaskScheduler(taskScheduler); + return _this(); + } + + /** + * The delay between checks of the file for new content in milliseconds. + * @param delay the delay. + * @return the spec. + * @see org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer#setPollingDelay(long) + */ + public TailAdapterSpec delay(long delay) { + this.factoryBean.setDelay(delay); + return _this(); + } + + /** + * The delay in milliseconds between attempts to tail a non-existent file, + * or between attempts to execute a process if it fails for any reason. + * @param fileDelay the fileDelay. + * @return the spec. + * @see FileTailingMessageProducerSupport#setTailAttemptsDelay(long) + */ + public TailAdapterSpec fileDelay(long fileDelay) { + this.factoryBean.setFileDelay(fileDelay); + return _this(); + } + + /** + * If {@code true}, tail from the end of the file, otherwise include all lines from the beginning. + * Default {@code true}. + * @param end the end. + * @return the spec. + * @see org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer#setEnd(boolean) + */ + public TailAdapterSpec end(boolean end) { + this.factoryBean.setEnd(end); + return _this(); + } + + /** + * If {@code true}, close and reopen the file between reading chunks. + * Default {@code false}. + * @param reopen the reopen. + * @return the spec. + * @see org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer#setReopen(boolean) + */ + public TailAdapterSpec reopen(boolean reopen) { + this.factoryBean.setReopen(reopen); + return _this(); + } + + @Override + public TailAdapterSpec id(String id) { + this.factoryBean.setBeanName(id); + return _this(); + } + + @Override + public TailAdapterSpec phase(int phase) { + this.factoryBean.setPhase(phase); + return _this(); + } + + @Override + public TailAdapterSpec autoStartup(boolean autoStartup) { + this.factoryBean.setAutoStartup(autoStartup); + return _this(); + } + + @Override + public TailAdapterSpec outputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + return _this(); + } + + @Override + public TailAdapterSpec errorChannel(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + return _this(); + } + + @Override + protected FileTailingMessageProducerSupport doGet() { + if (this.outputChannel == null) { + this.factoryBean.setOutputChannel(new NullChannel()); + } + FileTailingMessageProducerSupport tailingMessageProducerSupport = null; + try { + this.factoryBean.afterPropertiesSet(); + tailingMessageProducerSupport = this.factoryBean.getObject(); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + if (this.errorChannel != null) { + tailingMessageProducerSupport.setErrorChannel(this.errorChannel); + } + tailingMessageProducerSupport.setOutputChannel(this.outputChannel); + return tailingMessageProducerSupport; + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/package-info.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/package-info.java new file mode 100644 index 0000000000..4a447e5f7c --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides File Components support for Spring Integration Java DSL. + */ +package org.springframework.integration.file.dsl; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 8e74e50614..d655539409 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -205,11 +205,11 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply private final MessageSessionCallback messageSessionCallback; - private volatile ExpressionEvaluatingMessageProcessor renameProcessor = - new ExpressionEvaluatingMessageProcessor( - new SpelExpressionParser().parseExpression("headers." + FileHeaders.RENAME_TO)); + protected final Set