INT-4086: (S)FTP: Add Local File Tree Support
JIRA: https://jira.spring.io/browse/INT-4086 The remote files may be carried with sub-path, not just file name. The same sub-dirs logic can be achieve with the `localFilenameGeneratorExpression` * Add the logic into `AbstractInboundFileSynchronizer` to build sub-dirs for local files * Switch on `WatchService` for the internal `FileReadingMessageSource` in the `AbstractInboundFileSynchronizingMessageSource` to react for all the changes in the `localDirectory` Introduce `RecursiveDirectoryScanner` and use it for `AbstractInboundFileSynchronizingMessageSource` by default * Add `maxDepth` and `fileVisitOptions` options to the `RecursiveDirectoryScanner` * Polishing JavaDocs, tests and docs
This commit is contained in:
committed by
Gary Russell
parent
6dbc721d73
commit
96ca9647a4
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* Copyright 2002-2017 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.
|
||||
@@ -34,6 +34,7 @@ import org.springframework.messaging.MessagingException;
|
||||
* @author Iwein Fuld
|
||||
* @author Gunnar Hillert
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 2.0
|
||||
*/
|
||||
public class DefaultDirectoryScanner implements DirectoryScanner {
|
||||
@@ -80,12 +81,12 @@ public class DefaultDirectoryScanner implements DirectoryScanner {
|
||||
* @param file the file to try to claim.
|
||||
*/
|
||||
@Override
|
||||
public final boolean tryClaim(File file) {
|
||||
public boolean tryClaim(File file) {
|
||||
return (this.locker == null) || this.locker.lock(file);
|
||||
}
|
||||
|
||||
@Override
|
||||
public final List<File> listFiles(File directory) throws IllegalArgumentException {
|
||||
public List<File> listFiles(File directory) throws IllegalArgumentException {
|
||||
File[] files = listEligibleFiles(directory);
|
||||
if (files == null) {
|
||||
throw new MessagingException("The path [" + directory
|
||||
|
||||
@@ -273,6 +273,10 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
|
||||
this.useWatchService = useWatchService;
|
||||
}
|
||||
|
||||
public boolean isUseWatchService() {
|
||||
return this.useWatchService;
|
||||
}
|
||||
|
||||
/**
|
||||
* The {@link WatchService} event types.
|
||||
* If {@link #setUseWatchService} isn't {@code true}, this option is ignored.
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
/*
|
||||
* Copyright 2017 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;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.nio.file.FileVisitOption;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.springframework.integration.file.filters.AbstractFileListFilter;
|
||||
import org.springframework.integration.file.filters.FileListFilter;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* The {@link DefaultDirectoryScanner} extension which walks through the directory tree
|
||||
* using {@link Files#walk(Path, int, FileVisitOption...)}.
|
||||
* <p>
|
||||
* By default this class visits all levels of the file tree without any {@link FileVisitOption}s.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*
|
||||
* @see Files#walk
|
||||
*/
|
||||
public class RecursiveDirectoryScanner extends DefaultDirectoryScanner {
|
||||
|
||||
private int maxDepth = Integer.MAX_VALUE;
|
||||
|
||||
private FileVisitOption[] fileVisitOptions = new FileVisitOption[0];
|
||||
|
||||
/**
|
||||
* The maximum number of directory levels to visit.
|
||||
* @param maxDepth the maximum number of directory levels to visit
|
||||
*/
|
||||
public void setMaxDepth(int maxDepth) {
|
||||
this.maxDepth = maxDepth;
|
||||
}
|
||||
|
||||
/**
|
||||
* The options to configure the traversal.
|
||||
* @param fileVisitOptions options to configure the traversal
|
||||
*/
|
||||
public void setFileVisitOptions(FileVisitOption... fileVisitOptions) {
|
||||
Assert.notNull(fileVisitOptions, "'fileVisitOptions' must not be null");
|
||||
this.fileVisitOptions = fileVisitOptions;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<File> listFiles(File directory) throws IllegalArgumentException {
|
||||
FileListFilter<File> filter = getFilter();
|
||||
boolean supportAcceptFilter = filter instanceof AbstractFileListFilter;
|
||||
try {
|
||||
Stream<File> fileStream = Files.walk(directory.toPath(), this.maxDepth, this.fileVisitOptions)
|
||||
.skip(1)
|
||||
.map(Path::toFile)
|
||||
.filter(file -> !supportAcceptFilter
|
||||
|| ((AbstractFileListFilter<File>) filter).accept(file));
|
||||
|
||||
if (supportAcceptFilter) {
|
||||
return fileStream.collect(Collectors.toList());
|
||||
}
|
||||
else {
|
||||
return filter.filterFiles(fileStream.toArray(File[]::new));
|
||||
}
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new IllegalArgumentException(e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2016 the original author or authors.
|
||||
* Copyright 2002-2017 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.
|
||||
@@ -47,6 +47,6 @@ public abstract class AbstractFileListFilter<F> implements FileListFilter<F> {
|
||||
* @param file The file.
|
||||
* @return true if the file passes the filter.
|
||||
*/
|
||||
protected abstract boolean accept(F file);
|
||||
public abstract boolean accept(F file);
|
||||
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean accept(F file) {
|
||||
public boolean accept(F file) {
|
||||
String key = buildKey(file);
|
||||
synchronized (this.monitor) {
|
||||
String newValue = value(file);
|
||||
|
||||
@@ -60,7 +60,7 @@ public class ExpressionFileListFilter<F> extends AbstractFileListFilter<F>
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean accept(F file) {
|
||||
public boolean accept(F file) {
|
||||
return this.expression.getValue(getEvaluationContext(), file, Boolean.class);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015-2016 the original author or authors.
|
||||
* Copyright 2015-2017 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.
|
||||
@@ -31,7 +31,7 @@ public class IgnoreHiddenFileListFilter extends AbstractFileListFilter<File> {
|
||||
* @return Returns {@code true} for any non-hidden files.
|
||||
*/
|
||||
@Override
|
||||
protected boolean accept(File file) {
|
||||
public boolean accept(File file) {
|
||||
return !file.isHidden();
|
||||
}
|
||||
|
||||
|
||||
@@ -25,6 +25,7 @@ import java.io.OutputStream;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.regex.Matcher;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
@@ -100,7 +101,7 @@ public abstract class AbstractInboundFileSynchronizer<F>
|
||||
* Should we <em>transfer</em> the remote file <b>timestamp</b>
|
||||
* to the local file? By default this is false.
|
||||
*/
|
||||
private volatile boolean preserveTimestamp;
|
||||
private volatile boolean preserveTimestamp;
|
||||
|
||||
private BeanFactory beanFactory;
|
||||
|
||||
@@ -313,7 +314,12 @@ public abstract class AbstractInboundFileSynchronizer<F>
|
||||
long modified = getModified(remoteFile);
|
||||
|
||||
File localFile = new File(localDirectory, localFileName);
|
||||
if (!localFile.exists() || (this.preserveTimestamp && modified != localFile.lastModified())) {
|
||||
boolean exists = localFile.exists();
|
||||
if (!exists || (this.preserveTimestamp && modified != localFile.lastModified())) {
|
||||
if (!exists &&
|
||||
localFileName.replaceAll("/", Matcher.quoteReplacement(File.separator)).contains(File.separator)) {
|
||||
localFile.getParentFile().mkdirs(); //NOSONAR - will fail on the writing below
|
||||
}
|
||||
String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
|
||||
File tempFile = new File(tempFileName);
|
||||
OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
|
||||
|
||||
@@ -27,6 +27,7 @@ import org.springframework.beans.factory.BeanInitializationException;
|
||||
import org.springframework.context.Lifecycle;
|
||||
import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource;
|
||||
import org.springframework.integration.file.FileReadingMessageSource;
|
||||
import org.springframework.integration.file.RecursiveDirectoryScanner;
|
||||
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
|
||||
import org.springframework.integration.file.filters.CompositeFileListFilter;
|
||||
import org.springframework.integration.file.filters.FileListFilter;
|
||||
@@ -128,6 +129,22 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
|
||||
this.localFileListFilter = localFileListFilter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Switch the local {@link FileReadingMessageSource} to use its internal
|
||||
* {@code FileReadingMessageSource.WatchServiceDirectoryScanner}.
|
||||
* @param useWatchService the {@code boolean} flag to switch to
|
||||
* {@code FileReadingMessageSource.WatchServiceDirectoryScanner} on {@code true}.
|
||||
* @since 5.0
|
||||
*/
|
||||
public void setUseWatchService(boolean useWatchService) {
|
||||
this.fileSource.setUseWatchService(useWatchService);
|
||||
if (useWatchService) {
|
||||
this.fileSource.setWatchEvents(FileReadingMessageSource.WatchEventType.CREATE,
|
||||
FileReadingMessageSource.WatchEventType.MODIFY,
|
||||
FileReadingMessageSource.WatchEventType.DELETE);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
super.afterPropertiesSet();
|
||||
@@ -149,7 +166,15 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
|
||||
this.localFileListFilter = new FileSystemPersistentAcceptOnceFileListFilter(
|
||||
new SimpleMetadataStore(), getComponentName());
|
||||
}
|
||||
this.fileSource.setFilter(this.buildFilter());
|
||||
FileListFilter<File> filter = buildFilter();
|
||||
if (!this.fileSource.isUseWatchService()) {
|
||||
RecursiveDirectoryScanner directoryScanner = new RecursiveDirectoryScanner();
|
||||
directoryScanner.setFilter(filter);
|
||||
this.fileSource.setScanner(directoryScanner);
|
||||
}
|
||||
else {
|
||||
this.fileSource.setFilter(filter);
|
||||
}
|
||||
if (this.getBeanFactory() != null) {
|
||||
this.fileSource.setBeanFactory(this.getBeanFactory());
|
||||
}
|
||||
@@ -168,12 +193,14 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
|
||||
@Override
|
||||
public void start() {
|
||||
this.running = true;
|
||||
this.fileSource.start();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
this.running = false;
|
||||
try {
|
||||
this.fileSource.stop();
|
||||
this.synchronizer.close();
|
||||
}
|
||||
catch (IOException e) {
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
/*
|
||||
* Copyright 2017 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;
|
||||
|
||||
import static org.hamcrest.Matchers.hasItem;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.TemporaryFolder;
|
||||
|
||||
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
|
||||
|
||||
/**
|
||||
* @author Iwein Fuld
|
||||
* @author Gunnar Hillert
|
||||
* @author Gary Russell
|
||||
* @author Artem Bilan
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
public class RecursiveDirectoryScannerTests {
|
||||
|
||||
private File subFolder;
|
||||
|
||||
private File subSubFolder;
|
||||
|
||||
private File topLevelFile;
|
||||
|
||||
private File subLevelFile;
|
||||
|
||||
private File subSubLevelFile;
|
||||
|
||||
@Rule
|
||||
public TemporaryFolder recursivePath = new TemporaryFolder();
|
||||
|
||||
@Before
|
||||
public void setup() throws IOException {
|
||||
this.subFolder = this.recursivePath.newFolder("subFolder");
|
||||
this.subSubFolder = new File(this.subFolder, "subSubFolder");
|
||||
this.subSubFolder.mkdir();
|
||||
this.topLevelFile = this.recursivePath.newFile("file1");
|
||||
this.subLevelFile = new File(this.subFolder, "file2");
|
||||
this.subLevelFile.createNewFile();
|
||||
this.subSubLevelFile = new File(this.subSubFolder, "file3");
|
||||
this.subSubLevelFile.createNewFile();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldReturnAllFilesIncludingDirs() throws IOException {
|
||||
RecursiveDirectoryScanner scanner = new RecursiveDirectoryScanner();
|
||||
scanner.setFilter(new AcceptOnceFileListFilter<>());
|
||||
List<File> files = scanner.listFiles(this.recursivePath.getRoot());
|
||||
assertEquals(5, files.size());
|
||||
assertThat(files, hasItem(this.topLevelFile));
|
||||
assertThat(files, hasItem(this.subLevelFile));
|
||||
assertThat(files, hasItem(this.subSubLevelFile));
|
||||
assertThat(files, hasItem(this.subFolder));
|
||||
assertThat(files, hasItem(this.subSubFolder));
|
||||
File file = new File(this.subSubFolder, "file4");
|
||||
file.createNewFile();
|
||||
files = scanner.listFiles(this.recursivePath.getRoot());
|
||||
assertEquals(1, files.size());
|
||||
assertThat(files, hasItem(file));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -131,6 +131,7 @@ public class AbstractRemoteFileSynchronizerTests {
|
||||
source.setMaxFetchSize(1);
|
||||
source.setBeanName("maxFetchSizeSource");
|
||||
source.afterPropertiesSet();
|
||||
source.start();
|
||||
|
||||
source.receive();
|
||||
assertEquals(1, count.get());
|
||||
@@ -138,7 +139,7 @@ public class AbstractRemoteFileSynchronizerTests {
|
||||
source.receive();
|
||||
sync.synchronizeToLocalDirectory(mock(File.class), 1);
|
||||
source.receive();
|
||||
sync.close();
|
||||
source.stop();
|
||||
}
|
||||
|
||||
private AbstractInboundFileSynchronizer<String> createLimitingSynchronizer(final AtomicInteger count) {
|
||||
|
||||
Reference in New Issue
Block a user