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 index 196ebf16be..c220263623 100644 --- 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 @@ -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 private void listFiles() { String remoteDirectory = this.remoteDirectoryExpression.getValue(getEvaluationContext(), String.class); F[] files = this.remoteFileTemplate.list(remoteDirectory); - int maxFetchSize = getMaxFetchSize(); - List filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files); - if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) { - rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); - List 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> fileInfoList = asFileInfoList(filteredFiles); - Iterator> iterator = fileInfoList.iterator(); - while (iterator.hasNext()) { - AbstractFileInfo next = iterator.next(); - if (next.isDirectory()) { - iterator.remove(); + if (!ObjectUtils.isEmpty(files)) { + int maxFetchSize = getMaxFetchSize(); + List filteredFiles = this.filter == null ? Arrays.asList(files) : this.filter.filterFiles(files); + if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) { + rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); + List newList = new ArrayList<>(maxFetchSize); + for (int i = 0; i < maxFetchSize; i++) { + newList.add(filteredFiles.get(i)); + } + filteredFiles = newList; } - else { - next.setRemoteDirectory(remoteDirectory); + List> 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 filteredFiles, F file) { @@ -228,4 +225,6 @@ public abstract class AbstractRemoteFileStreamingMessageSource abstract protected List> asFileInfoList(Collection files); + abstract protected boolean isDirectory(F file); + } 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 index 364674af7c..b467659c1b 100644 --- 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 @@ -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 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 filteredFiles = filterFiles(files); if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { @@ -313,7 +317,6 @@ public abstract class AbstractInboundFileSynchronizer throw e1; } } - return copied; } else { @@ -344,7 +347,7 @@ public abstract class AbstractInboundFileSynchronizer ? (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); } 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 index 4badab662e..ac2bafb609 100644 --- 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 @@ -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 the file type. + * @return the array of remaining elements. + * @since 5.0.7 + */ + @SuppressWarnings("unchecked") + public static F[] purgeUnwantedElements(F[] fileArray, Predicate 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(); } 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 index e37c62871d..5a4c8b9f92 100644 --- 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 @@ -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 { 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 index e77d35d458..3b5898b9f2 100644 --- 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 @@ -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(); + } + } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpMessageSourceTests.java new file mode 100644 index 0000000000..7f7c78e98c --- /dev/null +++ b/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 { + + } + +} 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 index aba1a0af80..dc416c3323 100644 --- 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 @@ -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 received = (Message) 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 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 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 ftpMessageSource() { FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), Comparator.comparing(FileInfo::getFilename)); 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 index 5be1960c08..057c47441c 100644 --- 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 @@ -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(); + } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpMessageSourceTests.java new file mode 100644 index 0000000000..eb2142de59 --- /dev/null +++ b/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 { + + } + +} 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 index 6534cdf075..9a238eb5a1 100644 --- 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 @@ -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 received = (Message) 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 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 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 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 sftpMessageSource() { SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), Comparator.comparing(FileInfo::getFilename));