Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-8792: Fix File StreamingMS for one file filter

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/8792

1. Use a `SftpStreamingMessageSource` with a `maxFetchSize = 5` and a `ChainFileListFilter` filter composed with
`SftpSystemMarkerFilePresentFileListFilter` which `supportsSingleFileFiltering == false`
2. Put 2 files in the folder and invoke `SftpStreamingMessageSource.receive()` method twice.
3. Put 5 files in the folder and invoke `SftpStreamingMessageSource.receive()`` method five times.
4. The last two files won't be received.

When you set max fetch size to a number bigger than one (for example 5) and at a certain point
it is necessary to `this.toBeReceived.clear()` inside `AbstractRemoteFileStreamingMessageSource.doReceive()`,
those removed elements from toBeReceived are not rolled back.

* Fix `AbstractRemoteFileStreamingMessageSource.listFiles()` to calculate `maxFetchSize` as `getMaxFetchSize() - this.fetched.get()`

**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**

(cherry picked from commit 22c4db291b)
This commit is contained in:
Artem Bilan
2023-11-17 16:39:02 -05:00
parent fa298f8ec2
commit 17844869dc
2 changed files with 74 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -215,7 +215,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
if (this.filter != null && this.filter.supportsSingleFileFiltering()
&& !this.filter.accept(file.getFileInfo())) {
if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files
if (!this.toBeReceived.isEmpty()) { // don't re-fetch already filtered files
file = poll();
continue;
}
@@ -267,7 +267,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
}
protected AbstractFileInfo<F> poll() {
if (this.toBeReceived.size() == 0) {
if (this.toBeReceived.isEmpty()) {
listFiles();
}
return this.toBeReceived.poll();
@@ -297,7 +297,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
if (!ObjectUtils.isEmpty(files)) {
List<AbstractFileInfo<F>> fileInfoList;
if (this.filter != null && !this.filter.supportsSingleFileFiltering()) {
int maxFetchSize = getMaxFetchSize();
int maxFetchSize = getMaxFetchSize() - this.fetched.get();
List<F> filteredFiles = this.filter.filterFiles(files);
if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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,13 +16,17 @@
package org.springframework.integration.sftp.inbound;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.io.FileUtils;
import org.apache.sshd.sftp.client.SftpClient;
import org.junit.jupiter.api.Test;
@@ -39,11 +43,14 @@ import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.filters.AcceptAllFileListFilter;
import org.springframework.integration.file.filters.ChainFileListFilter;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.SftpTestSupport;
import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
import org.springframework.integration.sftp.filters.SftpSystemMarkerFilePresentFileListFilter;
import org.springframework.integration.sftp.session.SftpFileInfo;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.integration.transformer.StreamTransformer;
@@ -168,6 +175,69 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
StaticMessageHeaderAccessor.getCloseableResource(received).close();
}
@Test
public void maxFetchIsAdjustedWhenNoSupportsSingleFileFiltering() throws Exception {
SftpStreamingMessageSource messageSource = buildSource();
ChainFileListFilter<SftpClient.DirEntry> chainFileListFilter = new ChainFileListFilter<>();
SftpSystemMarkerFilePresentFileListFilter sftpSystemMarkerFilePresentFileListFilter =
new SftpSystemMarkerFilePresentFileListFilter(
new SftpSimplePatternFileListFilter("*"), ".trg");
SftpPersistentAcceptOnceFileListFilter sftpPersistentAcceptOnceFileListFilter =
new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "prefix");
chainFileListFilter.addFilter(sftpSystemMarkerFilePresentFileListFilter);
chainFileListFilter.addFilter(sftpPersistentAcceptOnceFileListFilter);
messageSource.setFilter(chainFileListFilter);
messageSource.setMaxFetchSize(5);
messageSource.afterPropertiesSet();
messageSource.start();
addFileAndTrigger("file001");
addFileAndTrigger("file002");
Message<InputStream> received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file001");
received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file002");
addFileAndTrigger("file003");
addFileAndTrigger("file004");
addFileAndTrigger("file005");
addFileAndTrigger("file006");
addFileAndTrigger("file007");
received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file003");
received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file004");
received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file005");
received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file006");
received = messageSource.receive();
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file007");
}
private void addFileAndTrigger(String filename) throws IOException {
File file = new File(this.sourceRemoteDirectory, filename);
FileUtils.writeStringToFile(file, "source1", StandardCharsets.UTF_8);
file = new File(this.sourceRemoteDirectory, filename + ".trg");
file.createNewFile();
}
private SftpStreamingMessageSource buildSource() {
SftpStreamingMessageSource messageSource =
new SftpStreamingMessageSource(this.config.template(),