INT-4496: (S)FTP: Sort file list earlier

garyrussell · artembilan · commit 11d39b7da119 · 2018-07-02T16:49:23.000-04:00
JIRA: https://jira.spring.io/browse/INT-4496 Sort the file list before applying `maxFetchSize` and filters. * Polishing - PR Comments
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
@@ -21,7 +21,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
@@ -59,7 +58,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 
 	private final BlockingQueue<AbstractFileInfo<F>> toBeReceived = new LinkedBlockingQueue<AbstractFileInfo<F>>();
 
-	private final Comparator<AbstractFileInfo<F>> comparator;
+	private final Comparator<F> comparator;
 
 	private boolean fileInfoJson = true;
 
@@ -76,7 +75,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 	private volatile FileListFilter<F> filter;
 
 	protected AbstractRemoteFileStreamingMessageSource(RemoteFileTemplate<F> template,
-			Comparator<AbstractFileInfo<F>> comparator) {
+			Comparator<F> comparator) {
 		this.remoteFileTemplate = template;
 		this.comparator = comparator;
 	}
@@ -194,7 +193,7 @@ private void listFiles() {
 		String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class);
 		F[] files = this.remoteFileTemplate.list(remoteDirectory);
 		if (!ObjectUtils.isEmpty(files)) {
-			files = FileUtils.purgeUnwantedElements(files, f -> f == null || isDirectory(f));
+			files = FileUtils.purgeUnwantedElements(files, f -> f == null || isDirectory(f), this.comparator);
 		}
 		if (!ObjectUtils.isEmpty(files)) {
 			int maxFetchSize = getMaxFetchSize();
@@ -209,9 +208,6 @@ private void listFiles() {
 			}
 			List<AbstractFileInfo<F>> fileInfoList = asFileInfoList(filteredFiles);
 			fileInfoList.forEach(fi -> fi.setRemoteDirectory(remoteDirectory));
-			if (this.comparator != null) {
-				Collections.sort(fileInfoList, this.comparator);
-			}
 			this.toBeReceived.addAll(fileInfoList);
 		}
 	}
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
@@ -24,6 +24,7 @@
 import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.List;
 import java.util.regex.Matcher;
 
@@ -59,6 +60,8 @@
  * {@link org.springframework.integration.file.filters.FileListFilter}s to
  * ensure the file entry is acceptable.
  *
+ * @param <F> the Type that represents a remote file.
+ *
  * @author Josh Long
  * @author Mark Fisher
  * @author Oleg Zhurakousky
@@ -116,6 +119,8 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 	private BeanFactory beanFactory;
 
+	private Comparator<F> comparator;
+
 	/**
 	 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
 	 *
@@ -127,6 +132,20 @@ public AbstractInboundFileSynchronizer(SessionFactory<F> sessionFactory) {
 	}
 
 
+	protected Comparator<F> getComparator() {
+		return this.comparator;
+	}
+
+	/**
+	 * Set a comparator to sort the retrieved list of {@code F} (the Type that represents
+	 * the remote file) prior to applying filters and max fetch size.
+	 * @param comparator the comparator.
+	 * @since 5.1
+	 */
+	public void setComparator(Comparator<F> comparator) {
+		this.comparator = comparator;
+	}
+
 	/**
 	 * @param remoteFileSeparator the remote file separator.
 	 * @see RemoteFileTemplate#setRemoteFileSeparator(String)
@@ -286,7 +305,7 @@ public void synchronizeToLocalDirectory(final File localDirectory, final int max
 			int transferred = this.remoteFileTemplate.execute(session -> {
 				F[] files = session.list(this.evaluatedRemoteDirectory);
 				if (!ObjectUtils.isEmpty(files)) {
-					files = FileUtils.purgeUnwantedElements(files, e -> !isFile(e));
+					files = FileUtils.purgeUnwantedElements(files, e -> !isFile(e), this.comparator);
 				}
 				if (!ObjectUtils.isEmpty(files)) {
 					List<F> filteredFiles = filterFiles(files);
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/support/FileUtils.java b/spring-integration-file/src/main/java/org/springframework/integration/file/support/FileUtils.java
@@ -19,8 +19,10 @@
 import java.lang.reflect.Array;
 import java.nio.file.FileSystems;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.function.Predicate;
 
+import org.springframework.lang.Nullable;
 import org.springframework.util.ObjectUtils;
 
 /**
@@ -38,19 +40,29 @@ public final class FileUtils {
 	 * Remove entries from the array if the predicate returns true for an element.
 	 * @param fileArray the array.
 	 * @param predicate the predicate.
+	 * @param comparator an optional comparator to sort the results.
 	 * @param <F> the file type.
 	 * @return the array of remaining elements.
 	 * @since 5.0.7
 	 */
 	@SuppressWarnings("unchecked")
-	public static <F> F[] purgeUnwantedElements(F[] fileArray, Predicate<F> predicate) {
+	public static <F> F[] purgeUnwantedElements(F[] fileArray, Predicate<F> predicate,
+			@Nullable Comparator<F> comparator) {
 		if (ObjectUtils.isEmpty(fileArray)) {
 			return fileArray;
 		}
 		else {
-			return Arrays.stream(fileArray)
-					.filter(predicate.negate())
-					.toArray(size -> (F[]) Array.newInstance(fileArray[0].getClass(), size));
+			if (comparator == null) {
+				return Arrays.stream(fileArray)
+						.filter(predicate.negate())
+						.toArray(size -> (F[]) Array.newInstance(fileArray[0].getClass(), size));
+			}
+			else {
+				return Arrays.stream(fileArray)
+						.filter(predicate.negate())
+						.sorted(comparator)
+						.toArray(size -> (F[]) Array.newInstance(fileArray[0].getClass(), size));
+			}
 		}
 	}
 
diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
@@ -192,7 +192,7 @@ public void testLineByLine() throws Exception {
 
 	public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {
 
-		protected Streamer(RemoteFileTemplate<String> template, Comparator<AbstractFileInfo<String>> comparator) {
+		protected Streamer(RemoteFileTemplate<String> template, Comparator<String> comparator) {
 			super(template, comparator);
 		}
 
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/Ftp.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/Ftp.java
@@ -21,7 +21,6 @@
 
 import org.apache.commons.net.ftp.FTPFile;
 
-import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.MessageSessionCallback;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
@@ -79,7 +78,7 @@ public static FtpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
 	 */
 	public static FtpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
 			RemoteFileTemplate<FTPFile> remoteFileTemplate,
-			Comparator<AbstractFileInfo<FTPFile>> receptionOrderComparator) {
+			Comparator<FTPFile> receptionOrderComparator) {
 		return new FtpStreamingInboundChannelAdapterSpec(remoteFileTemplate, receptionOrderComparator);
 	}
 
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,7 +23,6 @@
 import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
-import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
@@ -43,7 +42,7 @@ public class FtpStreamingInboundChannelAdapterSpec
 				FtpStreamingMessageSource> {
 
 	FtpStreamingInboundChannelAdapterSpec(RemoteFileTemplate<FTPFile> remoteFileTemplate,
-			Comparator<AbstractFileInfo<FTPFile>> comparator) {
+			Comparator<FTPFile> comparator) {
 		this.target = new FtpStreamingMessageSource(remoteFileTemplate, comparator);
 	}
 
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java
@@ -56,8 +56,7 @@ public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template) {
 	 * @param template the template.
 	 * @param comparator the comparator.
 	 */
-	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template,
-			Comparator<AbstractFileInfo<FTPFile>> comparator) {
+	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template, Comparator<FTPFile> comparator) {
 		super(template, comparator);
 		doSetFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpStreamingMessageSource"));
 	}
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
@@ -46,7 +46,6 @@
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AcceptAllFileListFilter;
-import org.springframework.integration.file.remote.FileInfo;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.FtpTestSupport;
 import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
@@ -160,7 +159,7 @@ public void testMaxFetchNoFilter() throws IOException {
 
 	private FtpStreamingMessageSource buildSource() {
 		FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(this.config.template(),
-				Comparator.comparing(FileInfo::getFilename));
+				Comparator.comparing(FTPFile::getName));
 		messageSource.setRemoteDirectory("ftpSource/");
 		messageSource.setMaxFetchSize(1);
 		messageSource.setBeanFactory(this.context);
@@ -193,11 +192,10 @@ public ConcurrentMap<String, String> metadataMap() {
 		@InboundChannelAdapter(channel = "stream", autoStartup = "false")
 		public MessageSource<InputStream> ftpMessageSource() {
 			FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(),
-					Comparator.comparing(FileInfo::getFilename));
+					Comparator.comparing(FTPFile::getName));
 			messageSource.setFilter(
 					new FtpPersistentAcceptOnceFileListFilter(
 							new SimpleMetadataStore(metadataMap()), "testStreaming"));
-
 			messageSource.setRemoteDirectory("ftpSource/");
 			return messageSource;
 		}
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/Sftp.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/Sftp.java
@@ -19,7 +19,6 @@
 import java.io.File;
 import java.util.Comparator;
 
-import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.MessageSessionCallback;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
@@ -80,7 +79,7 @@ public static SftpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
 	 */
 	public static SftpStreamingInboundChannelAdapterSpec inboundStreamingAdapter(
 			RemoteFileTemplate<LsEntry> remoteFileTemplate,
-			Comparator<AbstractFileInfo<LsEntry>> receptionOrderComparator) {
+			Comparator<LsEntry> receptionOrderComparator) {
 		return new SftpStreamingInboundChannelAdapterSpec(remoteFileTemplate, receptionOrderComparator);
 	}
 
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java
@@ -21,7 +21,6 @@
 import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
-import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
@@ -41,7 +40,7 @@ public class SftpStreamingInboundChannelAdapterSpec
 		SftpStreamingMessageSource> {
 
 	SftpStreamingInboundChannelAdapterSpec(RemoteFileTemplate<LsEntry> remoteFileTemplate,
-			Comparator<AbstractFileInfo<LsEntry>> comparator) {
+			Comparator<LsEntry> comparator) {
 		this.target = new SftpStreamingMessageSource(remoteFileTemplate, comparator);
 	}
 
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java
@@ -57,7 +57,7 @@ public SftpStreamingMessageSource(RemoteFileTemplate<LsEntry> template) {
 	 * @param comparator the comparator.
 	 */
 	public SftpStreamingMessageSource(RemoteFileTemplate<LsEntry> template,
-			Comparator<AbstractFileInfo<LsEntry>> comparator) {
+			Comparator<LsEntry> comparator) {
 		super(template, comparator);
 		doSetFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "sftpStreamingMessageSource"));
 	}
diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
@@ -43,7 +43,6 @@
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AcceptAllFileListFilter;
-import org.springframework.integration.file.remote.FileInfo;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.sftp.SftpTestSupport;
@@ -161,7 +160,7 @@ public void testMaxFetchLambdaFilter() throws IOException {
 
 	private SftpStreamingMessageSource buildSource() {
 		SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(this.config.template(),
-				Comparator.comparing(FileInfo::getFilename));
+				Comparator.comparing(LsEntry::getFilename));
 		messageSource.setRemoteDirectory("sftpSource/");
 		messageSource.setMaxFetchSize(1);
 		messageSource.setBeanFactory(this.context);
@@ -189,7 +188,7 @@ public PollerMetadata defaultPoller() {
 		@InboundChannelAdapter(channel = "stream", autoStartup = "false")
 		public MessageSource<InputStream> sftpMessageSource() {
 			SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(),
-					Comparator.comparing(FileInfo::getFilename));
+					Comparator.comparing(LsEntry::getFilename));
 			messageSource.setFilter(new AcceptAllFileListFilter<>());
 			messageSource.setRemoteDirectory("sftpSource/");
 			return messageSource;
diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc
@@ -567,6 +567,8 @@ If you set the `fileInfoJson` property on the `FtpStreamingMessageSource` to `fa
 The `FTPFile` object provided by the underlying Apache Net library can be accessed using the `FtpFileInfo.getFileInfo()` method.
 The `fileInfoJson` property is not available when using XML configuration but you can set it by injecting the `FtpStreamingMessageSource` into one of your configuration classes.
 
+Starting with version 5.1, the `comparator` 's generic type is now `FTPFile`; previously it was `AbstractFileInfo<FTPFile>`; this is because the sort is now performed earlier in the processing, before filtering and applying `maxFetch`.
+
 ==== Configuring with Java Configuration
 
 The following Spring Boot application provides an example of configuring the inbound adapter using Java configuration:
@@ -727,6 +729,8 @@ Another use for `max-fetch-size` is if you want to stop fetching remote files, b
 Setting the `maxFetchSize` property on the `MessageSource` (programmatically, via JMX, or via a <<control-bus, control bus>>) effectively stops the adapter from fetching more files, but allows the poller to continue to emit messages for files that have previously been fetched.
 If the poller is active when the property is changed, the change will take effect on the next poll.
 
+Starting with version 5.1, the synchronizer can be provided with a `Comparator<FTPFile>`.
+This is useful when restricting the number of files fetched with `maxFetchSize`.
 
 [[ftp-outbound]]
 === FTP Outbound Channel Adapter
diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc
@@ -612,6 +612,8 @@ If you set the `fileInfoJson` property on the `SftpStreamingMessageSource` to `f
 The `LsEntry` object provided by the underlying Jsch library can be accessed using the `SftpFileInfo.getFileInfo()` method.
 The `fileInfoJson` property is not available when using XML configuration but you can set it by injecting the `SftpStreamingMessageSource` into one of your configuration classes.
 
+Starting with version 5.1, the `comparator` 's generic type is now `LsEntry`; previously it was `AbstractFileInfo<LsEntry>`; this is because the sort is now performed earlier in the processing, before filtering and applying `maxFetch`.
+
 ==== Configuring with Java Configuration
 
 The following Spring Boot application provides an example of configuring the inbound adapter using Java configuration:
@@ -772,6 +774,8 @@ Another use for `max-fetch-size` is if you want to stop fetching remote files, b
 Setting the `maxFetchSize` property on the `MessageSource` (programmatically, via JMX, or via a <<control-bus, control bus>>) effectively stops the adapter from fetching more files, but allows the poller to continue to emit messages for files that have previously been fetched.
 If the poller is active when the property is changed, the change will take effect on the next poll.
 
+Starting with version 5.1, the synchronizer can be provided with a `Comparator<LsEntry>`.
+This is useful when restricting the number of files fetched with `maxFetchSize`.
 
 [[sftp-outbound]]
 === SFTP Outbound Channel Adapter
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
@@ -85,3 +85,8 @@ A `RotatingServerAdvice` is now available to poll multiple servers and/or direct
 See <<ftp-rotating-server-advice>> and <<sftp-rotating-server-advice>> for more information.
 
 Also inbound adapter `localFilenameExpression` s can contain the variable `#remoteDirectory` which contains the remote directory being polled.
+
+The generic type of the comparators, used to sort the fetched file list for the streaming adapters, has changed from `Comparator<AbstractFileInfo<F>>` to simply `Comparator<F>`.
+See <<ftp-streaming>> and <<sftp-streaming>> for more information.
+
+In addition, the synchronizers for inbound channel adapters can now be provided with a `Comparator`; this is useful when using `maxFetchSize` to limit the files retrieved.

Original file line number	Diff line number	Diff line change
`@@ -192,7 +192,7 @@ public void testLineByLine() throws Exception {`
`192`	`192`
`193`	`193`	`public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {`
`194`	`194`
`195`		`- protected Streamer(RemoteFileTemplate<String> template, Comparator<AbstractFileInfo<String>> comparator) {`
	`195`	`+ protected Streamer(RemoteFileTemplate<String> template, Comparator<String> comparator) {`
`196`	`196`	`super(template, comparator);`
`197`	`197`	`}`
`198`	`198`