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
This commit is contained in:
lsgeme
@@ -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<F>
|
||||
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<F> 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<F> poll() {
|
||||
if (this.toBeReceived.size() == 0) {
|
||||
listFiles();
|
||||
}
|
||||
return this.toBeReceived.poll();
|
||||
}
|
||||
|
||||
private void resetFilterIfNecessary(AbstractFileInfo<F> file) {
|
||||
@@ -237,13 +244,6 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
|
||||
}
|
||||
}
|
||||
|
||||
protected AbstractFileInfo<F> poll() {
|
||||
if (this.toBeReceived.size() == 0) {
|
||||
listFiles();
|
||||
}
|
||||
return this.toBeReceived.poll();
|
||||
}
|
||||
|
||||
protected String remotePath(AbstractFileInfo<F> file) {
|
||||
return file.getRemoteDirectory().endsWith(this.remoteFileSeparator)
|
||||
? file.getRemoteDirectory() + file.getFilename()
|
||||
|
||||
@@ -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<String> remoteFileTemplate = mock(RemoteFileTemplate.class);
|
||||
when(remoteFileTemplate.list("remoteDirectory")).thenReturn(new String[] { "file1", "file2" });
|
||||
Session<String> session = mock(Session.class);
|
||||
when(session.readRaw(anyString())).thenReturn(mock(InputStream.class));
|
||||
when(remoteFileTemplate.getSession()).thenReturn(session);
|
||||
|
||||
FileListFilter<String> fileListFilter = mock(FileListFilter.class);
|
||||
when(fileListFilter.supportsSingleFileFiltering()).thenReturn(true);
|
||||
when(fileListFilter.accept("file1")).thenReturn(false);
|
||||
when(fileListFilter.accept("file2")).thenReturn(false);
|
||||
|
||||
Comparator<String> 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<String> {
|
||||
|
||||
TestRemoteFileStreamingMessageSource(RemoteFileTemplate<String> template, Comparator<String> comparator) {
|
||||
super(template, comparator);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected List<AbstractFileInfo<String>> asFileInfoList(Collection<String> 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<String> {
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user