INT-4495: (S)FTP: fix max-fetch with directories

JIRA: https://jira.spring.io/browse/INT-4495

Previously, `maxFetch` was applied before directories were removed from
the fetch list.

Remove the directories before filtering and applying `maxFetch`.

* Polishing - PR Comments

* More polishing

* More polishing.

* Check for empty array.

* Remove test main method.

spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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 java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
-import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -38,8 +37,10 @@ import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.filters.ReversibleFileListFilter;
 import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.file.support.FileUtils;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * A message source that produces a message with an {@link InputStream} payload
@@ -192,31 +193,27 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 	private void listFiles() {
 		String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class);
 		F[] files = this.remoteFileTemplate.list(remoteDirectory);
-		int maxFetchSize = getMaxFetchSize();
-		List<F> filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files);
-		if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
-			rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
-			List<F> newList = new ArrayList<>(maxFetchSize);
-			for (int i = 0; i < maxFetchSize; i++) {
-				newList.add(filteredFiles.get(i));
-			}
-			filteredFiles = newList;
+		if (!ObjectUtils.isEmpty(files)) {
+			files = FileUtils.purgeUnwantedElements(files, f -> f == null || isDirectory(f));
 		}
-		List<AbstractFileInfo<F>> fileInfoList = asFileInfoList(filteredFiles);
-		Iterator<AbstractFileInfo<F>> iterator = fileInfoList.iterator();
-		while (iterator.hasNext()) {
-			AbstractFileInfo<F> next = iterator.next();
-			if (next.isDirectory()) {
-				iterator.remove();
+		if (!ObjectUtils.isEmpty(files)) {
+			int maxFetchSize = getMaxFetchSize();
+			List<F> filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files);
+			if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
+				rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
+				List<F> newList = new ArrayList<>(maxFetchSize);
+				for (int i = 0; i < maxFetchSize; i++) {
+					newList.add(filteredFiles.get(i));
+				}
+				filteredFiles = newList;
 			}
-			else {
-				next.setRemoteDirectory(remoteDirectory);
+			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);
 		}
-		if (this.comparator != null) {
-			Collections.sort(fileInfoList, this.comparator);
-		}
-		this.toBeReceived.addAll(fileInfoList);
 	}
 
 	protected void rollbackFromFileToListEnd(List<F> filteredFiles, F file) {
@@ -228,4 +225,6 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 
 	abstract protected List<AbstractFileInfo<F>> asFileInfoList(Collection<F> files);
 
+	abstract protected boolean isDirectory(F file);
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java

@@ -46,6 +46,7 @@ import org.springframework.integration.file.filters.ReversibleFileListFilter;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileUtils;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -284,6 +285,9 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		try {
 			int transferred = this.remoteFileTemplate.execute(session -> {
 				F[] files = session.list(this.evaluatedRemoteDirectory);
+				if (!ObjectUtils.isEmpty(files)) {
+					files = FileUtils.purgeUnwantedElements(files, e -> !isFile(e));
+				}
 				if (!ObjectUtils.isEmpty(files)) {
 					List<F> filteredFiles = filterFiles(files);
 					if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) {
@@ -313,7 +317,6 @@ public abstract class AbstractInboundFileSynchronizer<F>
 							throw e1;
 						}
 					}
-
 					return copied;
 				}
 				else {
@@ -344,7 +347,7 @@ public abstract class AbstractInboundFileSynchronizer<F>
 				? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
 				: remoteFileName;
 
-		if (!this.isFile(remoteFile)) {
+		if (!isFile(remoteFile)) {
 			if (this.logger.isDebugEnabled()) {
 				this.logger.debug("cannot copy, not a file: " + remoteFilePath);
 			}

spring-integration-file/src/main/java/org/springframework/integration/file/support/FileUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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.
@@ -16,7 +16,12 @@
 
 package org.springframework.integration.file.support;
 
+import java.lang.reflect.Array;
 import java.nio.file.FileSystems;
+import java.util.Arrays;
+import java.util.function.Predicate;
+
+import org.springframework.util.ObjectUtils;
 
 /**
  * Utilities for operations on Files.
@@ -29,6 +34,26 @@ public final class FileUtils {
 
 	public static final boolean IS_POSIX = FileSystems.getDefault().supportedFileAttributeViews().contains("posix");
 
+	/**
+	 * Remove entries from the array if the predicate returns true for an element.
+	 * @param fileArray the array.
+	 * @param predicate the predicate.
+	 * @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) {
+		if (ObjectUtils.isEmpty(fileArray)) {
+			return fileArray;
+		}
+		else {
+			return Arrays.stream(fileArray)
+					.filter(predicate.negate())
+					.toArray(size -> (F[]) Array.newInstance(fileArray[0].getClass(), size));
+		}
+	}
+
 	private FileUtils() {
 		super();
 	}

spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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.
@@ -210,6 +210,11 @@ public class StreamingInboundTests {
 			return infos;
 		}
 
+		@Override
+		protected boolean isDirectory(String file) {
+			return false;
+		}
+
 	}
 
 	public static class StringFileInfo extends AbstractFileInfo<String> {

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -76,4 +76,9 @@ public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessag
 		return canonicalFiles;
 	}
 
+	@Override
+	protected boolean isDirectory(FTPFile file) {
+		return file != null && file.isDirectory();
+	}
+
 }

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpMessageSourceTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 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.
+ * 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.ftp.inbound;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.ftp.FtpTestSupport;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Gary Russell
+ * @since 5.0.7
+ *
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class FtpMessageSourceTests extends FtpTestSupport {
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void testMaxFetch() throws Exception {
+		FtpInboundFileSynchronizingMessageSource messageSource = buildSource();
+		Message<?> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.FILENAME), equalTo(" ftpSource1.txt"));
+	}
+
+	private FtpInboundFileSynchronizingMessageSource buildSource() throws Exception {
+		FtpInboundFileSynchronizer sync = new FtpInboundFileSynchronizer(sessionFactory());
+		sync.setRemoteDirectory("ftpSource/");
+		sync.setBeanFactory(this.context);
+		FtpInboundFileSynchronizingMessageSource messageSource = new FtpInboundFileSynchronizingMessageSource(sync);
+		messageSource.setLocalDirectory(getTargetLocalDirectory());
+		messageSource.setMaxFetchSize(1);
+		messageSource.setBeanFactory(this.context);
+		messageSource.setBeanName("source");
+		messageSource.afterPropertiesSet();
+		return messageSource;
+	}
+
+	@Configuration
+	public static class Config {
+
+	}
+
+}

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java

@@ -31,6 +31,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.InboundChannelAdapter;
@@ -52,7 +53,7 @@ import org.springframework.integration.transformer.StreamTransformer;
 import org.springframework.messaging.Message;
 import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Gary Russell
@@ -61,7 +62,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * @since 4.3
  *
  */
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 
@@ -74,6 +75,12 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 	@Autowired
 	private SourcePollingChannelAdapter adapter;
 
+	@Autowired
+	private Config config;
+
+	@Autowired
+	private ApplicationContext context;
+
 	@Rule
 	public Log4j2LevelAdjuster adjuster =
 			Log4j2LevelAdjuster.debug()
@@ -82,6 +89,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testAllContents() {
+		this.adapter.start();
 		Message<byte[]> received = (Message<byte[]>) this.data.receive(10000);
 		assertNotNull(received);
 		assertThat(new String(received.getPayload()), equalTo("source1"));
@@ -115,6 +123,35 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		this.adapter.stop();
 	}
 
+	@Test
+	public void testMaxFetch() {
+		FtpStreamingMessageSource messageSource = buildsource();
+		messageSource.setFilter(new AcceptAllFileListFilter<>());
+		messageSource.afterPropertiesSet();
+		Message<InputStream> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" ftpSource1.txt"));
+	}
+
+	@Test
+	public void testMaxFetchNoFilter() {
+		FtpStreamingMessageSource messageSource = buildsource();
+		messageSource.setFilter(null);
+		messageSource.afterPropertiesSet();
+		Message<InputStream> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" ftpSource1.txt"));
+	}
+
+	private FtpStreamingMessageSource buildsource() {
+		FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(this.config.template(),
+				Comparator.comparing(FileInfo::getFilename));
+		messageSource.setRemoteDirectory("ftpSource/");
+		messageSource.setMaxFetchSize(1);
+		messageSource.setBeanFactory(this.context);
+		return messageSource;
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class Config {
@@ -133,7 +170,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		}
 
 		@Bean
-		@InboundChannelAdapter(channel = "stream")
+		@InboundChannelAdapter(channel = "stream", autoStartup = "false")
 		public MessageSource<InputStream> ftpMessageSource() {
 			FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(),
 					Comparator.comparing(FileInfo::getFilename));

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -76,4 +76,9 @@ public class SftpStreamingMessageSource extends AbstractRemoteFileStreamingMessa
 		return canonicalFiles;
 	}
 
+	@Override
+	protected boolean isDirectory(LsEntry file) {
+		return file != null && file.getAttrs().isDir();
+	}
+
 }

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpMessageSourceTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 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.
+ * 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.sftp.inbound;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.sftp.SftpTestSupport;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Gary Russell
+ * @since 5.0.7
+ *
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class SftpMessageSourceTests extends SftpTestSupport {
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void testMaxFetch() throws Exception {
+		SftpInboundFileSynchronizingMessageSource messageSource = buildSource();
+		Message<?> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.FILENAME), equalTo(" sftpSource1.txt"));
+	}
+
+	private SftpInboundFileSynchronizingMessageSource buildSource() throws Exception {
+		SftpInboundFileSynchronizer sync = new SftpInboundFileSynchronizer(sessionFactory());
+		sync.setRemoteDirectory("sftpSource/");
+		sync.setBeanFactory(this.context);
+		SftpInboundFileSynchronizingMessageSource messageSource = new SftpInboundFileSynchronizingMessageSource(sync);
+		messageSource.setLocalDirectory(getTargetLocalDirectory());
+		messageSource.setMaxFetchSize(1);
+		messageSource.setBeanFactory(this.context);
+		messageSource.setBeanName("source");
+		messageSource.afterPropertiesSet();
+		return messageSource;
+	}
+
+	@Configuration
+	public static class Config {
+
+	}
+
+}

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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,12 +23,14 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 
 import java.io.InputStream;
+import java.util.Arrays;
 import java.util.Comparator;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.InboundChannelAdapter;
@@ -49,7 +51,7 @@ import org.springframework.integration.transformer.StreamTransformer;
 import org.springframework.messaging.Message;
 import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
 
@@ -59,7 +61,7 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
  * @since 4.3
  *
  */
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 
@@ -72,9 +74,16 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 	@Autowired
 	private SourcePollingChannelAdapter adapter;
 
+	@Autowired
+	private Config config;
+
+	@Autowired
+	private ApplicationContext context;
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testAllContents() {
+		this.adapter.start();
 		Message<byte[]> received = (Message<byte[]>) this.data.receive(10000);
 		assertNotNull(received);
 		assertThat(new String(received.getPayload()), equalTo("source1"));
@@ -108,6 +117,45 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		this.adapter.stop();
 	}
 
+	@Test
+	public void testMaxFetch() {
+		SftpStreamingMessageSource messageSource = buildSource();
+		messageSource.setFilter(new AcceptAllFileListFilter<>());
+		messageSource.afterPropertiesSet();
+		Message<InputStream> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" sftpSource1.txt"));
+	}
+
+	@Test
+	public void testMaxFetchNoFilter() {
+		SftpStreamingMessageSource messageSource = buildSource();
+		messageSource.setFilter(null);
+		messageSource.afterPropertiesSet();
+		Message<InputStream> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" sftpSource1.txt"));
+	}
+
+	@Test
+	public void testMaxFetchLambdaFilter() {
+		SftpStreamingMessageSource messageSource = buildSource();
+		messageSource.setFilter(f -> Arrays.asList(f));
+		messageSource.afterPropertiesSet();
+		Message<InputStream> received = messageSource.receive();
+		assertNotNull(received);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" sftpSource1.txt"));
+	}
+
+	private SftpStreamingMessageSource buildSource() {
+		SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(this.config.template(),
+				Comparator.comparing(FileInfo::getFilename));
+		messageSource.setRemoteDirectory("sftpSource/");
+		messageSource.setMaxFetchSize(1);
+		messageSource.setBeanFactory(this.context);
+		return messageSource;
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class Config {
@@ -126,7 +174,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		}
 
 		@Bean
-		@InboundChannelAdapter(channel = "stream")
+		@InboundChannelAdapter(channel = "stream", autoStartup = "false")
 		public MessageSource<InputStream> sftpMessageSource() {
 			SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(),
 					Comparator.comparing(FileInfo::getFilename));