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
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
507764a3d6
commit
f71d6a0e66
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<FileInboundChannelAdapterSpec, FileReadingMessageSource> {
|
||||
|
||||
private final FileListFilterFactoryBean fileListFilterFactoryBean = new FileListFilterFactoryBean();
|
||||
|
||||
private FileLocker locker;
|
||||
|
||||
FileInboundChannelAdapterSpec() {
|
||||
this.target = new FileReadingMessageSource();
|
||||
}
|
||||
|
||||
FileInboundChannelAdapterSpec(Comparator<File> 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
|
||||
* <em>true</em>. If set to <em>false</em> 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<File> 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<FileSplitterSpec, FileSplitter> {
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 <F> the target file type.
|
||||
* @param <S> the target {@link FileTransferringMessageHandlerSpec} implementation type.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransferringMessageHandlerSpec<F, S>>
|
||||
extends MessageHandlerSpec<S, FileTransferringMessageHandler<F>>
|
||||
implements ComponentsRegistration {
|
||||
|
||||
private FileNameGenerator fileNameGenerator;
|
||||
|
||||
private DefaultFileNameGenerator defaultFileNameGenerator;
|
||||
|
||||
protected FileTransferringMessageHandlerSpec(SessionFactory<F> sessionFactory) {
|
||||
this.target = new FileTransferringMessageHandler<>(sessionFactory);
|
||||
}
|
||||
|
||||
protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTemplate) {
|
||||
this.target = new FileTransferringMessageHandler<>(remoteFileTemplate);
|
||||
}
|
||||
|
||||
protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> 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 <P> the expected payload type.
|
||||
* @return the current Spec
|
||||
*/
|
||||
public <P> S remoteDirectory(Function<Message<P>, 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 <P> the expected payload type.
|
||||
* @return the current Spec
|
||||
*/
|
||||
public <P> S temporaryRemoteDirectory(Function<Message<P>, 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<Object> getComponentsToRegister() {
|
||||
if (this.defaultFileNameGenerator != null) {
|
||||
return Collections.singletonList(this.defaultFileNameGenerator);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<FileWritingMessageHandlerSpec, FileWritingMessageHandler>
|
||||
implements ComponentsRegistration {
|
||||
|
||||
private FileNameGenerator fileNameGenerator;
|
||||
|
||||
private DefaultFileNameGenerator defaultFileNameGenerator;
|
||||
|
||||
FileWritingMessageHandlerSpec(File destinationDirectory) {
|
||||
this.target = new FileWritingMessageHandler(destinationDirectory);
|
||||
}
|
||||
|
||||
FileWritingMessageHandlerSpec(String directoryExpression) {
|
||||
this(PARSER.parseExpression(directoryExpression));
|
||||
}
|
||||
|
||||
<P> FileWritingMessageHandlerSpec(Function<Message<P>, ?> 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
|
||||
* <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.
|
||||
* @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 <em>false</em>. When set to <em>true</em>, 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<Object> getComponentsToRegister() {
|
||||
if (this.defaultFileNameGenerator != null) {
|
||||
return Collections.singletonList(this.defaultFileNameGenerator);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<File> 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 <P> the payload type.
|
||||
* @return the {@link FileWritingMessageHandlerSpec} instance.
|
||||
*/
|
||||
public static <P> FileWritingMessageHandlerSpec outboundAdapter(Function<Message<P>, ?> 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 <P> the payload type.
|
||||
* @return the {@link FileWritingMessageHandlerSpec} instance.
|
||||
*/
|
||||
public static <P> FileWritingMessageHandlerSpec outboundGateway(Function<Message<P>, ?> 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 <em>false</em>.
|
||||
* @return the {@link FileToByteArrayTransformer}.
|
||||
*/
|
||||
public static FileToByteArrayTransformer toByteArrayTransformer(boolean deleteFiles) {
|
||||
FileToByteArrayTransformer fileToByteArrayTransformer = new FileToByteArrayTransformer();
|
||||
fileToByteArrayTransformer.setDeleteFiles(deleteFiles);
|
||||
return fileToByteArrayTransformer;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 <F> the target file type.
|
||||
* @param <S> the target {@link RemoteFileInboundChannelAdapterSpec} implementation type.
|
||||
* @param <MS> the target {@link AbstractInboundFileSynchronizingMessageSource} implementation type.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFileInboundChannelAdapterSpec<F, S, MS>,
|
||||
MS extends AbstractInboundFileSynchronizingMessageSource<F>>
|
||||
extends MessageSourceSpec<S, MS> implements ComponentsRegistration {
|
||||
|
||||
protected final AbstractInboundFileSynchronizer<F> synchronizer;
|
||||
|
||||
private CompositeFileListFilter<F> filter;
|
||||
|
||||
protected RemoteFileInboundChannelAdapterSpec(AbstractInboundFileSynchronizer<F> 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<File> 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<String, String> 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<F> filter) {
|
||||
if (this.filter == null) {
|
||||
if (filter instanceof CompositeFileListFilter) {
|
||||
this.filter = (CompositeFileListFilter<F>) filter;
|
||||
}
|
||||
else {
|
||||
this.filter = new CompositeFileListFilter<F>();
|
||||
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<Object> getComponentsToRegister() {
|
||||
return Collections.singletonList(this.synchronizer);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 <F> the target file type.
|
||||
* @param <S> the target {@link RemoteFileOutboundGatewaySpec} implementation type.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutboundGatewaySpec<F, S>>
|
||||
extends MessageHandlerSpec<S, AbstractRemoteFileOutboundGateway<F>> {
|
||||
|
||||
private CompositeFileListFilter<F> filter;
|
||||
|
||||
private CompositeFileListFilter<File> mputFilter;
|
||||
|
||||
protected RemoteFileOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<F> 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 <P> the expected payload type.
|
||||
* @return the spec
|
||||
*/
|
||||
public <P> S localDirectory(Function<Message<P>, 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<F> filter) {
|
||||
if (this.filter == null) {
|
||||
if (filter instanceof CompositeFileListFilter) {
|
||||
this.filter = (CompositeFileListFilter<F>) 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 <em>local</em> file system view when
|
||||
* using {@code MPUT} command.
|
||||
* @param filter the filter to set
|
||||
* @return the Spec.
|
||||
*/
|
||||
public S mputFilter(FileListFilter<File> filter) {
|
||||
if (this.mputFilter == null) {
|
||||
if (filter instanceof CompositeFileListFilter) {
|
||||
this.mputFilter = (CompositeFileListFilter<File>) 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 <em>local</em> 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 <em>local</em> 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 <P> the expected payload type.
|
||||
* @return the Spec.
|
||||
*/
|
||||
public <P> S renameFunction(Function<Message<P>, 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 <P> the expected payload type.
|
||||
* @return the Spec.
|
||||
*/
|
||||
public <P> S localFilename(Function<Message<P>, 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);
|
||||
|
||||
}
|
||||
@@ -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<TailAdapterSpec, FileTailingMessageProducerSupport> {
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
/**
|
||||
* Provides File Components support for Spring Integration Java DSL.
|
||||
*/
|
||||
package org.springframework.integration.file.dsl;
|
||||
@@ -205,11 +205,11 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
|
||||
private final MessageSessionCallback<F, ?> messageSessionCallback;
|
||||
|
||||
private volatile ExpressionEvaluatingMessageProcessor<String> renameProcessor =
|
||||
new ExpressionEvaluatingMessageProcessor<String>(
|
||||
new SpelExpressionParser().parseExpression("headers." + FileHeaders.RENAME_TO));
|
||||
protected final Set<Option> options = new HashSet<>();
|
||||
|
||||
protected volatile Set<Option> options = new HashSet<Option>();
|
||||
private volatile ExpressionEvaluatingMessageProcessor<String> renameProcessor =
|
||||
new ExpressionEvaluatingMessageProcessor<>(
|
||||
new SpelExpressionParser().parseExpression("headers." + FileHeaders.RENAME_TO));
|
||||
|
||||
private volatile Expression localDirectoryExpression;
|
||||
|
||||
@@ -313,20 +313,36 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify the array of options for various gateway commands.
|
||||
* @param options the options to set
|
||||
*/
|
||||
public void setOptions(String options) {
|
||||
String[] opts = options.split("\\s");
|
||||
for (String opt : opts) {
|
||||
String trimmedOpt = opt.trim();
|
||||
if (StringUtils.hasLength(trimmedOpt)) {
|
||||
this.options.add(Option.toOption(trimmedOpt));
|
||||
}
|
||||
}
|
||||
Assert.hasText(options, "'options' must not be empty.");
|
||||
this.options.clear();
|
||||
Arrays.stream(options.split("\\s"))
|
||||
.filter(StringUtils::hasText)
|
||||
.map(s -> Option.toOption(s.trim()))
|
||||
.forEach(this.options::add);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param remoteFileSeparator the remoteFileSeparator to set
|
||||
* Specify the array of options for various gateway commands.
|
||||
* @param options the {@link Option} array to use.
|
||||
* @since 5.0
|
||||
*/
|
||||
public void setOptions(Option... options) {
|
||||
Assert.notNull(options, "'options' must not be null");
|
||||
Assert.noNullElements(options, "'options' cannot contain null element");
|
||||
|
||||
this.options.clear();
|
||||
|
||||
Collections.addAll(this.options, options);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the file separator when dealing with remote files; default '/'.
|
||||
* @param remoteFileSeparator the separator.
|
||||
* @see RemoteFileTemplate#setRemoteFileSeparator(String)
|
||||
*/
|
||||
public void setRemoteFileSeparator(String remoteFileSeparator) {
|
||||
@@ -334,6 +350,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a directory path where remote files will be transferred to.
|
||||
* @param localDirectory the localDirectory to set
|
||||
*/
|
||||
public void setLocalDirectory(File localDirectory) {
|
||||
@@ -342,11 +359,17 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a SpEL expression to evaluate directory path where remote files will be transferred to.
|
||||
* @param localDirectoryExpression the SpEL to determine the local directory.
|
||||
*/
|
||||
public void setLocalDirectoryExpression(Expression localDirectoryExpression) {
|
||||
this.localDirectoryExpression = localDirectoryExpression;
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@code boolean} flag to identify if local directory should be created automatically.
|
||||
* Defaults to {@code true}.
|
||||
* @param autoCreateLocalDirectory the autoCreateLocalDirectory to set
|
||||
*/
|
||||
public void setAutoCreateLocalDirectory(boolean autoCreateLocalDirectory) {
|
||||
@@ -354,6 +377,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the temporary suffix to use when transferring files to the remote system.
|
||||
* Default {@code .writing}.
|
||||
* @param temporaryFileSuffix the temporaryFileSuffix to set
|
||||
* @see RemoteFileTemplate#setTemporaryFileSuffix(String)
|
||||
*/
|
||||
@@ -362,6 +387,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a {@link FileListFilter} to filter remote files.
|
||||
* @param filter the filter to set
|
||||
*/
|
||||
public void setFilter(FileListFilter<F> filter) {
|
||||
@@ -369,6 +395,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link FileListFilter} that runs against the <em>local</em> file system view when
|
||||
* using {@code MPUT} command.
|
||||
* @param filter the filter to set
|
||||
*/
|
||||
public void setMputFilter(FileListFilter<File> filter) {
|
||||
@@ -376,6 +404,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a SpEL expression for files renaming during transfer.
|
||||
* @param renameExpression the expression to use.
|
||||
* @since 4.3
|
||||
*/
|
||||
@@ -384,6 +413,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a SpEL expression for files renaming during transfer.
|
||||
* @param renameExpression the String in SpEL syntax.
|
||||
* @since 4.3
|
||||
*/
|
||||
@@ -393,6 +423,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a SpEL expression for local files renaming after downloading.
|
||||
* @param localFilenameGeneratorExpression the expression to use.
|
||||
* @since 3.0
|
||||
*/
|
||||
@@ -402,6 +433,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a SpEL expression for local files renaming after downloading.
|
||||
* @param localFilenameGeneratorExpression the String in SpEL syntax.
|
||||
* @since 4.3
|
||||
*/
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
package org.springframework.integration.file.remote.handler;
|
||||
|
||||
import org.springframework.expression.Expression;
|
||||
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.session.SessionFactory;
|
||||
@@ -62,6 +63,7 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
|
||||
|
||||
/**
|
||||
* A {@code boolean} flag to indicate automatically create the directory or not.
|
||||
* @param autoCreateDirectory true to automatically create the directory.
|
||||
* @see RemoteFileTemplate#setAutoCreateDirectory(boolean)
|
||||
*/
|
||||
@@ -70,6 +72,7 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a remote file separator symbol.
|
||||
* @param remoteFileSeparator the remote file separator.
|
||||
* @see RemoteFileTemplate#setRemoteFileSeparator(String)
|
||||
*/
|
||||
@@ -78,6 +81,7 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a remote directory path SpEL expression.
|
||||
* @param remoteDirectoryExpression the remote directory expression
|
||||
* @see RemoteFileTemplate#setRemoteDirectoryExpression(Expression)
|
||||
*/
|
||||
@@ -86,6 +90,7 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a remote directory path SpEL expression.
|
||||
* @param temporaryRemoteDirectoryExpression the temporary remote directory expression
|
||||
* @see RemoteFileTemplate#setTemporaryRemoteDirectoryExpression(Expression)
|
||||
*/
|
||||
@@ -102,6 +107,8 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@code boolean} flag to use temporary files names or not.
|
||||
* Defaults to {@code true}.
|
||||
* @param useTemporaryFileName true to use a temporary file name.
|
||||
* @see RemoteFileTemplate#setUseTemporaryFileName(boolean)
|
||||
*/
|
||||
@@ -110,6 +117,8 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* @see RemoteFileTemplate#setFileNameGenerator(FileNameGenerator)
|
||||
*/
|
||||
@@ -118,6 +127,8 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* @see RemoteFileTemplate#setCharset(String)
|
||||
*/
|
||||
@@ -126,6 +137,8 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the temporary suffix to use when transferring files to the remote system.
|
||||
* Default ".writing".
|
||||
* @param temporaryFileSuffix the temporary file suffix.
|
||||
* @see RemoteFileTemplate#setTemporaryFileSuffix(String)
|
||||
*/
|
||||
|
||||
@@ -185,10 +185,6 @@ public abstract class AbstractInboundFileSynchronizer<F>
|
||||
this.preserveTimestamp = preserveTimestamp;
|
||||
}
|
||||
|
||||
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
|
||||
this.evaluationContext = evaluationContext;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
|
||||
this.beanFactory = beanFactory;
|
||||
|
||||
@@ -84,7 +84,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
|
||||
super.doStart();
|
||||
destroyProcess();
|
||||
this.command = "tail " + this.options + " " + this.getFile().getAbsolutePath();
|
||||
this.getTaskExecutor().execute(() -> runExec());
|
||||
this.getTaskExecutor().execute(this::runExec);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -170,8 +170,8 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
|
||||
if (logger.isInfoEnabled()) {
|
||||
logger.info("Restarting tail process in " + getMissingFileDelay() + " milliseconds");
|
||||
}
|
||||
getRequiredTaskScheduler().schedule((Runnable) () -> runExec(),
|
||||
new Date(System.currentTimeMillis() + getMissingFileDelay()));
|
||||
getRequiredTaskScheduler()
|
||||
.schedule(this::runExec, new Date(System.currentTimeMillis() + getMissingFileDelay()));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -0,0 +1,440 @@
|
||||
/*
|
||||
* 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 static org.hamcrest.Matchers.endsWith;
|
||||
import static org.hamcrest.Matchers.instanceOf;
|
||||
import static org.junit.Assert.assertArrayEquals;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.FileOutputStream;
|
||||
import java.nio.charset.Charset;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.TemporaryFolder;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.aop.TargetSource;
|
||||
import org.springframework.aop.framework.Advised;
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.beans.factory.ListableBeanFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
|
||||
import org.springframework.context.Lifecycle;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.ComponentScan;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.annotation.IntegrationComponentScan;
|
||||
import org.springframework.integration.annotation.MessagingGateway;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.IntegrationFlowDefinition;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.dsl.Pollers;
|
||||
import org.springframework.integration.dsl.StandardIntegrationFlow;
|
||||
import org.springframework.integration.dsl.channel.MessageChannels;
|
||||
import org.springframework.integration.file.DefaultFileNameGenerator;
|
||||
import org.springframework.integration.file.FileHeaders;
|
||||
import org.springframework.integration.file.FileReadingMessageSource;
|
||||
import org.springframework.integration.file.splitter.FileSplitter;
|
||||
import org.springframework.integration.file.support.FileExistsMode;
|
||||
import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessageHandlingException;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.messaging.support.ErrorMessage;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
import org.springframework.util.StreamUtils;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@DirtiesContext
|
||||
public class FileTests {
|
||||
|
||||
@ClassRule
|
||||
public static final TemporaryFolder tmpDir = new TemporaryFolder();
|
||||
|
||||
@Autowired
|
||||
private ListableBeanFactory beanFactory;
|
||||
|
||||
@Autowired
|
||||
private ControlBusGateway controlBus;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileFlow1Input")
|
||||
private MessageChannel fileFlow1Input;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileWriting.handler")
|
||||
private MessageHandler fileWritingMessageHandler;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("tailChannel")
|
||||
private PollableChannel tailChannel;
|
||||
|
||||
@Autowired
|
||||
private ApacheCommonsFileTailingMessageProducer tailer;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileReadingResultChannel")
|
||||
private PollableChannel fileReadingResultChannel;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileWritingInput")
|
||||
private MessageChannel fileWritingInput;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileWritingResultChannel")
|
||||
private PollableChannel fileWritingResultChannel;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileSplittingResultChannel")
|
||||
private PollableChannel fileSplittingResultChannel;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileTriggerFlow.input")
|
||||
private MessageChannel fileTriggerFlowInput;
|
||||
|
||||
@Autowired
|
||||
private CountDownLatch flushPredicateCalled;
|
||||
|
||||
@Test
|
||||
public void testFileHandler() throws Exception {
|
||||
Message<?> message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build();
|
||||
try {
|
||||
this.fileFlow1Input.send(message);
|
||||
fail("NullPointerException expected");
|
||||
}
|
||||
catch (Exception e) {
|
||||
assertThat(e, instanceOf(MessageHandlingException.class));
|
||||
assertThat(e.getCause(), instanceOf(NullPointerException.class));
|
||||
}
|
||||
DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
|
||||
fileNameGenerator.setBeanFactory(this.beanFactory);
|
||||
Object targetFileWritingMessageHandler = this.fileWritingMessageHandler;
|
||||
if (this.fileWritingMessageHandler instanceof Advised) {
|
||||
TargetSource targetSource = ((Advised) this.fileWritingMessageHandler).getTargetSource();
|
||||
if (targetSource != null) {
|
||||
targetFileWritingMessageHandler = targetSource.getTarget();
|
||||
}
|
||||
}
|
||||
DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler);
|
||||
dfa.setPropertyValue("fileNameGenerator", fileNameGenerator);
|
||||
this.fileFlow1Input.send(message);
|
||||
|
||||
assertTrue(new File(tmpDir.getRoot(), "foo").exists());
|
||||
|
||||
this.fileTriggerFlowInput.send(new GenericMessage<>("trigger"));
|
||||
assertTrue(this.flushPredicateCalled.await(10, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageProducerFlow() throws Exception {
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir.getRoot(), "TailTest"));
|
||||
for (int i = 0; i < 50; i++) {
|
||||
file.write((i + "\n").getBytes());
|
||||
}
|
||||
this.tailer.start();
|
||||
for (int i = 0; i < 50; i++) {
|
||||
Message<?> message = this.tailChannel.receive(5000);
|
||||
assertNotNull(message);
|
||||
assertEquals("hello " + i, message.getPayload());
|
||||
}
|
||||
assertNull(this.tailChannel.receive(1));
|
||||
|
||||
this.controlBus.send("@tailer.stop()");
|
||||
file.close();
|
||||
}
|
||||
|
||||
@Autowired
|
||||
private PollableChannel filePollingErrorChannel;
|
||||
|
||||
@Test
|
||||
public void testFileReadingFlow() throws Exception {
|
||||
List<Integer> evens = new ArrayList<>(25);
|
||||
for (int i = 0; i < 50; i++) {
|
||||
boolean even = i % 2 == 0;
|
||||
String extension = even ? ".sitest" : ".foofile";
|
||||
if (even) {
|
||||
evens.add(i);
|
||||
}
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir.getRoot(), i + extension));
|
||||
file.write(("" + i).getBytes());
|
||||
file.flush();
|
||||
file.close();
|
||||
}
|
||||
|
||||
Message<?> message = fileReadingResultChannel.receive(60000);
|
||||
assertNotNull(message);
|
||||
Object payload = message.getPayload();
|
||||
assertThat(payload, instanceOf(List.class));
|
||||
@SuppressWarnings("unchecked")
|
||||
List<String> result = (List<String>) payload;
|
||||
assertEquals(25, result.size());
|
||||
result.forEach(s -> assertTrue(evens.contains(Integer.parseInt(s))));
|
||||
|
||||
new File(tmpDir.getRoot(), "a.sitest").createNewFile();
|
||||
Message<?> receive = this.filePollingErrorChannel.receive(60000);
|
||||
assertNotNull(receive);
|
||||
assertThat(receive, instanceOf(ErrorMessage.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFileWritingFlow() throws Exception {
|
||||
String payload = "Spring Integration";
|
||||
this.fileWritingInput.send(new GenericMessage<>(payload));
|
||||
Message<?> receive = this.fileWritingResultChannel.receive(1000);
|
||||
assertNotNull(receive);
|
||||
assertThat(receive.getPayload(), instanceOf(File.class));
|
||||
File resultFile = (File) receive.getPayload();
|
||||
assertThat(resultFile.getAbsolutePath(),
|
||||
endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.write")));
|
||||
String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset());
|
||||
assertEquals(payload, fileContent);
|
||||
}
|
||||
|
||||
@Autowired
|
||||
@Qualifier("fileSplitter.handler")
|
||||
private MessageHandler fileSplitter;
|
||||
|
||||
@Test
|
||||
public void testFileSplitterFlow() throws Exception {
|
||||
FileOutputStream file = new FileOutputStream(new File(tmpDir.getRoot(), "foo.tmp"));
|
||||
file.write(("HelloWorld\näöüß").getBytes(Charset.defaultCharset()));
|
||||
file.flush();
|
||||
file.close();
|
||||
|
||||
Message<?> receive = this.fileSplittingResultChannel.receive(10000);
|
||||
assertNotNull(receive);
|
||||
assertThat(receive.getPayload(), instanceOf(FileSplitter.FileMarker.class)); // FileMarker.Mark.START
|
||||
assertEquals(0, receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE));
|
||||
receive = this.fileSplittingResultChannel.receive(10000);
|
||||
assertNotNull(receive); //HelloWorld
|
||||
receive = this.fileSplittingResultChannel.receive(10000);
|
||||
assertNotNull(receive); //äöüß
|
||||
receive = this.fileSplittingResultChannel.receive(10000);
|
||||
assertNotNull(receive);
|
||||
assertThat(receive.getPayload(), instanceOf(FileSplitter.FileMarker.class)); // FileMarker.Mark.END
|
||||
assertNull(this.fileSplittingResultChannel.receive(1));
|
||||
|
||||
assertEquals(StandardCharsets.US_ASCII, TestUtils.getPropertyValue(this.fileSplitter, "charset"));
|
||||
}
|
||||
|
||||
@Autowired
|
||||
@Qualifier("dynamicAdaptersResult")
|
||||
PollableChannel dynamicAdaptersResult;
|
||||
|
||||
@Autowired
|
||||
private MyService myService;
|
||||
|
||||
@Test
|
||||
public void testDynamicFileFlows() throws Exception {
|
||||
File newFolder1 = tmpDir.newFolder();
|
||||
FileOutputStream file = new FileOutputStream(new File(newFolder1, "foo"));
|
||||
file.write(("foo").getBytes());
|
||||
file.flush();
|
||||
file.close();
|
||||
|
||||
File newFolder2 = tmpDir.newFolder();
|
||||
file = new FileOutputStream(new File(newFolder2, "bar"));
|
||||
file.write(("bar").getBytes());
|
||||
file.flush();
|
||||
file.close();
|
||||
|
||||
this.myService.pollDirectories(newFolder1, newFolder2);
|
||||
|
||||
Set<String> payloads = new TreeSet<>();
|
||||
Message<?> receive = this.dynamicAdaptersResult.receive(10000);
|
||||
assertNotNull(receive);
|
||||
payloads.add((String) receive.getPayload());
|
||||
receive = this.dynamicAdaptersResult.receive(10000);
|
||||
assertNotNull(receive);
|
||||
payloads.add((String) receive.getPayload());
|
||||
|
||||
assertArrayEquals(new String[] { "bar", "foo" }, payloads.toArray());
|
||||
}
|
||||
|
||||
@MessagingGateway(defaultRequestChannel = "controlBus.input")
|
||||
private interface ControlBusGateway {
|
||||
|
||||
void send(String command);
|
||||
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableIntegration
|
||||
@ComponentScan
|
||||
@IntegrationComponentScan
|
||||
public static class ContextConfiguration {
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow controlBus() {
|
||||
return IntegrationFlowDefinition::controlBus;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow fileTriggerFlow() {
|
||||
return f -> f.handle("fileWriting.handler", "trigger");
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CountDownLatch flushPredicateCalled() {
|
||||
return new CountDownLatch(1);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow fileFlow1() {
|
||||
return IntegrationFlows.from("fileFlow1Input")
|
||||
.handle(Files.outboundAdapter(tmpDir.getRoot())
|
||||
.fileNameGenerator(message -> null)
|
||||
.fileExistsMode(FileExistsMode.APPEND_NO_FLUSH)
|
||||
.flushPredicate((fileAbsolutePath, lastWrite, filterMessage) -> {
|
||||
flushPredicateCalled().countDown();
|
||||
return true;
|
||||
}),
|
||||
c -> c.id("fileWriting"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow tailFlow() {
|
||||
return IntegrationFlows
|
||||
.from(Files.tailAdapter(new File(tmpDir.getRoot(), "TailTest"))
|
||||
.delay(500)
|
||||
.end(false)
|
||||
.id("tailer")
|
||||
.autoStartup(false))
|
||||
.transform("hello "::concat)
|
||||
.channel(MessageChannels.queue("tailChannel"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow fileReadingFlow() {
|
||||
return IntegrationFlows
|
||||
.from(Files.inboundAdapter(tmpDir.getRoot())
|
||||
.patternFilter("*.sitest")
|
||||
.useWatchService(true)
|
||||
.watchEvents(FileReadingMessageSource.WatchEventType.CREATE,
|
||||
FileReadingMessageSource.WatchEventType.MODIFY),
|
||||
e -> e.poller(Pollers.fixedDelay(100)
|
||||
.errorChannel("filePollingErrorChannel")))
|
||||
.filter(File.class, p -> !p.getName().startsWith("a"),
|
||||
e -> e.throwExceptionOnRejection(true))
|
||||
.transform(Files.toStringTransformer())
|
||||
.aggregate(a -> a.correlationExpression("1")
|
||||
.releaseStrategy(g -> g.size() == 25))
|
||||
.channel(MessageChannels.queue("fileReadingResultChannel"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public PollableChannel filePollingErrorChannel() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow fileWritingFlow() {
|
||||
return IntegrationFlows.from("fileWritingInput")
|
||||
.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.write")
|
||||
.header("directory", new File(tmpDir.getRoot(), "fileWritingFlow")))
|
||||
.handle(Files.outboundGateway(m -> m.getHeaders().get("directory")))
|
||||
.channel(MessageChannels.queue("fileWritingResultChannel"))
|
||||
.get();
|
||||
}
|
||||
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow fileSplitterFlow() {
|
||||
return IntegrationFlows
|
||||
.from(Files.inboundAdapter(tmpDir.getRoot())
|
||||
.patternFilter("foo.tmp"),
|
||||
e -> e.poller(p -> p.fixedDelay(100)))
|
||||
.split(Files.splitter()
|
||||
.markers()
|
||||
.charset(StandardCharsets.US_ASCII)
|
||||
.applySequence(true),
|
||||
e -> e.id("fileSplitter"))
|
||||
.channel(c -> c.queue("fileSplittingResultChannel"))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public PollableChannel dynamicAdaptersResult() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Service
|
||||
public static class MyService {
|
||||
|
||||
@Autowired
|
||||
private AutowireCapableBeanFactory beanFactory;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("dynamicAdaptersResult")
|
||||
PollableChannel dynamicAdaptersResult;
|
||||
|
||||
void pollDirectories(File... directories) {
|
||||
for (File directory : directories) {
|
||||
StandardIntegrationFlow integrationFlow = IntegrationFlows
|
||||
.from(Files.inboundAdapter(directory),
|
||||
e -> e.poller(p -> p.fixedDelay(1000))
|
||||
.id(directory.getName() + ".adapter"))
|
||||
.transform(Files.toStringTransformer(),
|
||||
e -> e.id(directory.getName() + ".transformer"))
|
||||
.channel(this.dynamicAdaptersResult)
|
||||
.get();
|
||||
this.beanFactory.initializeBean(integrationFlow, directory.getName());
|
||||
this.beanFactory.getBean(directory.getName() + ".transformer", Lifecycle.class).start();
|
||||
this.beanFactory.getBean(directory.getName() + ".adapter", Lifecycle.class).start();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -178,13 +178,11 @@ public class ScriptsTests {
|
||||
assertThat(payload, Matchers.instanceOf(Date.class));
|
||||
|
||||
// Some time window to avoid dates collision
|
||||
Thread.sleep(500);
|
||||
Thread.sleep(2);
|
||||
|
||||
assertTrue(((Date) payload).before(new Date()));
|
||||
|
||||
assertNotNull(this.messageSourceChannel.receive(20000));
|
||||
|
||||
assertNull(this.messageSourceChannel.receive(10));
|
||||
}
|
||||
|
||||
@Configuration
|
||||
|
||||
Reference in New Issue
Block a user