From 1820c05c26ad625d449d49b867a72f8d1cf97571 Mon Sep 17 00:00:00 2001 From: lsgeme Date: Thu, 7 Nov 2019 18:23:44 +0100 Subject: [PATCH] GH-3105: Fix (S)FTP streaming single filter logic Fixes https://github.com/spring-projects/spring-integration/issues/3105 The loop in the `AbstractRemoteFileStreamingMessageSource` doesn't check the next polled file for filtering. * Add `continue;` when we filter the current file and poll the next one. So, we go over into the `while()` beginning * Clean up the code style and cover more code path in test --- ...tractRemoteFileStreamingMessageSource.java | 72 ++++----- ...RemoteFileStreamingMessageSourceTests.java | 143 ++++++++++++++++++ 2 files changed, 179 insertions(+), 36 deletions(-) create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java 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 4ba84d7b8e..57a326ffa0 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 @@ -49,6 +49,7 @@ import org.springframework.util.ObjectUtils; * * @author Gary Russell * @author Artem Bilan + * @author Lukas Gemela * * @since 4.3 * @@ -180,51 +181,57 @@ public abstract class AbstractRemoteFileStreamingMessageSource return this.running.get(); } + @Override + protected Object doReceive(int maxFetchSize) { + return doReceive(); + } + @Override protected Object doReceive() { Assert.state(this.running.get(), () -> getComponentName() + " is not running"); AbstractFileInfo file = poll(); while (file != null) { if (this.filter != null && this.filter.supportsSingleFileFiltering() - && !this.filter.accept(file.getFileInfo())) { + && !this.filter.accept(file.getFileInfo())) { - if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files - file = poll(); - } - else { - file = null; - } + if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files + file = poll(); + continue; + } + else { + break; + } } - if (file != null) { + try { + String remotePath = remotePath(file); + Session session = this.remoteFileTemplate.getSession(); try { - String remotePath = remotePath(file); - Session session = this.remoteFileTemplate.getSession(); - try { - return getMessageBuilderFactory() - .withPayload(session.readRaw(remotePath)) - .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) - .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) - .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) - .setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort()) - .setHeader(FileHeaders.REMOTE_FILE_INFO, - this.fileInfoJson ? file.toJson() : file); - } - catch (IOException e) { - throw new UncheckedIOException("IOException when retrieving " + remotePath, e); - } + return getMessageBuilderFactory() + .withPayload(session.readRaw(remotePath)) + .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session) + .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory()) + .setHeader(FileHeaders.REMOTE_FILE, file.getFilename()) + .setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort()) + .setHeader(FileHeaders.REMOTE_FILE_INFO, + this.fileInfoJson ? file.toJson() : file); } - catch (RuntimeException e) { - resetFilterIfNecessary(file); - throw e; + catch (IOException e) { + throw new UncheckedIOException("IOException when retrieving " + remotePath, e); } } + catch (RuntimeException e) { + resetFilterIfNecessary(file); + throw e; + } } return null; } - @Override - protected Object doReceive(int maxFetchSize) { - return doReceive(); + protected AbstractFileInfo poll() { + if (this.toBeReceived.size() == 0) { + listFiles(); + } + return this.toBeReceived.poll(); } private void resetFilterIfNecessary(AbstractFileInfo file) { @@ -237,13 +244,6 @@ public abstract class AbstractRemoteFileStreamingMessageSource } } - protected AbstractFileInfo poll() { - if (this.toBeReceived.size() == 0) { - listFiles(); - } - return this.toBeReceived.poll(); - } - protected String remotePath(AbstractFileInfo file) { return file.getRemoteDirectory().endsWith(this.remoteFileSeparator) ? file.getRemoteDirectory() + file.getFilename() diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java new file mode 100644 index 0000000000..6eb530a89c --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java @@ -0,0 +1,143 @@ +/* + * Copyright 2015-2019 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 + * + * https://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.remote; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.session.Session; + +/** + * @author Lukas Gemela + * @author Artem Bilan + * + * @since 5.2.2 + * + */ +public class RemoteFileStreamingMessageSourceTests { + + @Test + @SuppressWarnings("unchecked") + public void filterOutFilesNotAcceptedByFilter() throws IOException { + RemoteFileTemplate remoteFileTemplate = mock(RemoteFileTemplate.class); + when(remoteFileTemplate.list("remoteDirectory")).thenReturn(new String[] { "file1", "file2" }); + Session session = mock(Session.class); + when(session.readRaw(anyString())).thenReturn(mock(InputStream.class)); + when(remoteFileTemplate.getSession()).thenReturn(session); + + FileListFilter fileListFilter = mock(FileListFilter.class); + when(fileListFilter.supportsSingleFileFiltering()).thenReturn(true); + when(fileListFilter.accept("file1")).thenReturn(false); + when(fileListFilter.accept("file2")).thenReturn(false); + + Comparator comparator = mock(Comparator.class); + TestRemoteFileStreamingMessageSource testRemoteFileStreamingMessageSource = + new TestRemoteFileStreamingMessageSource(remoteFileTemplate, comparator); + + testRemoteFileStreamingMessageSource.setFilter(fileListFilter); + testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory"); + testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class)); + testRemoteFileStreamingMessageSource.start(); + + assertThat(testRemoteFileStreamingMessageSource.doReceive()).isNull(); + } + + static class TestRemoteFileStreamingMessageSource extends AbstractRemoteFileStreamingMessageSource { + + TestRemoteFileStreamingMessageSource(RemoteFileTemplate template, Comparator comparator) { + super(template, comparator); + } + + @Override + protected List> asFileInfoList(Collection files) { + return files + .stream() + .map(TestFileInfo::new) + .collect(Collectors.toList()); + } + + @Override + protected boolean isDirectory(String file) { + return false; + } + + @Override + public String getComponentType() { + return null; + } + + } + + static class TestFileInfo extends AbstractFileInfo { + + TestFileInfo(String fileName) { + this.fileName = fileName; + } + + private final String fileName; + + @Override + public boolean isDirectory() { + return false; + } + + @Override + public boolean isLink() { + return false; + } + + @Override + public long getSize() { + return 0; + } + + @Override + public long getModified() { + return 0; + } + + @Override + public String getFilename() { + return fileName; + } + + @Override + public String getPermissions() { + return null; + } + + @Override + public String getFileInfo() { + return null; + } + + } + +}