From f3d61ec44a92dae5f71bb2e8351a4d79ff72222d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 1 Jul 2020 11:18:00 -0400 Subject: [PATCH] GH-3315: Fix (S)FTP Stream with Fair Rotation Resolves https://github.com/spring-projects/spring-integration/issues/3315 `maxFetchSize` ignored with filters supporting single file filtering (default). This breaks "fair" rotation with the `RotatingServerAdvice`. Honor `maxFetchSize`, even with filters that support single file filtering. **cherry-pick to 5.3.x, 5.2.x** --- ...tractRemoteFileStreamingMessageSource.java | 17 ++- ...RemoteFileStreamingMessageSourceTests.java | 7 +- .../ftp/inbound/RotatingServersTests.java | 122 +++++++++++++++++- 3 files changed, 131 insertions(+), 15 deletions(-) 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 f33eeb06b5..fd9fc2b2b5 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-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -27,6 +27,7 @@ import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.springframework.context.Lifecycle; import org.springframework.expression.Expression; @@ -65,6 +66,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource private final AtomicBoolean running = new AtomicBoolean(); + private final AtomicInteger fetched = new AtomicInteger(); + private boolean fileInfoJson = true; /** @@ -183,12 +186,11 @@ public abstract class AbstractRemoteFileStreamingMessageSource @Override protected Object doReceive(int maxFetchSize) { - return doReceive(); - } - - @Override - protected Object doReceive() { Assert.state(this.running.get(), () -> getComponentName() + " is not running"); + if (maxFetchSize > 0 && this.fetched.get() >= maxFetchSize) { + this.toBeReceived.clear(); + this.fetched.set(0); + } AbstractFileInfo file = poll(); while (file != null) { if (this.filter != null && this.filter.supportsSingleFileFiltering() @@ -205,6 +207,9 @@ public abstract class AbstractRemoteFileStreamingMessageSource try { String remotePath = remotePath(file); Session session = this.remoteFileTemplate.getSession(); + if (maxFetchSize > 0) { + this.fetched.incrementAndGet(); + } try { return getMessageBuilderFactory() .withPayload(session.readRaw(remotePath)) 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 index df17233b5e..631a765f68 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -18,6 +18,7 @@ package org.springframework.integration.file.remote; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -70,7 +71,7 @@ public class RemoteFileStreamingMessageSourceTests { testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class)); testRemoteFileStreamingMessageSource.start(); - assertThat(testRemoteFileStreamingMessageSource.doReceive()).isNull(); + assertThat(testRemoteFileStreamingMessageSource.doReceive(-1)).isNull(); } @Test @@ -94,7 +95,7 @@ public class RemoteFileStreamingMessageSourceTests { testRemoteFileStreamingMessageSource.start(); assertThatExceptionOfType(UncheckedIOException.class) - .isThrownBy(testRemoteFileStreamingMessageSource::doReceive); + .isThrownBy(() -> testRemoteFileStreamingMessageSource.doReceive(anyInt())); assertThat(cachingSessionFactory.getSession()).isNotNull(); } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java index 57a321d637..bc11430437 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java @@ -37,13 +37,15 @@ import org.junit.jupiter.api.TestInfo; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.MessageChannels; import org.springframework.integration.dsl.Pollers; -import org.springframework.integration.dsl.StandardIntegrationFlow; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.remote.aop.RotatingServerAdvice; import org.springframework.integration.file.remote.aop.RotationPolicy; import org.springframework.integration.file.remote.session.CachingSessionFactory; @@ -58,6 +60,7 @@ import org.springframework.integration.ftp.dsl.Ftp; import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.messaging.Message; /** * @author Gary Russell @@ -88,18 +91,49 @@ public class RotatingServersTests extends FtpTestSupport { }); } + @BeforeEach + public void extraSetup(TestInfo info) { + if (info.getTestMethod().get().getName().equals("testFairStreaming")) { + FtpRemoteFileTemplate rft = new FtpRemoteFileTemplate(sessionFactory()); + rft.execute(s -> { + ByteArrayInputStream bais = new ByteArrayInputStream("foo".getBytes()); + // 2 files per server, remove empty dirs + s.write(bais, "foo/f4"); + s.write(bais, "baz/f5"); + s.write(bais, "fiz/f6"); + s.rmdir("bar"); + s.rmdir("qux"); + s.rmdir("buz"); + return null; + }); + } + } + @BeforeEach @AfterEach public void clean(TestInfo info) { recursiveDelete(new File(tmpDir), info); } + @AfterEach + public void extraCleanUp(TestInfo info) { + if (info.getTestMethod().get().getName().equals("testFairStreaming")) { + FtpRemoteFileTemplate rft = new FtpRemoteFileTemplate(sessionFactory()); + rft.execute(s -> { + s.remove("foo/f4"); + s.remove("baz/f5"); + s.remove("fiz/f6"); + return null; + }); + } + } + @Test public void testStandard() throws Exception { AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StandardConfig.class); StandardConfig config = ctx.getBean(StandardConfig.class); - ctx.getBean(StandardIntegrationFlow.class).stop(); assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + ctx.getBean(SourcePollingChannelAdapter.class).stop(); List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1); File f1 = new File(tmpDir + File.separator + "standard" + File.separator + "f1"); @@ -116,8 +150,8 @@ public class RotatingServersTests extends FtpTestSupport { public void testFair() throws Exception { AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(FairConfig.class); StandardConfig config = ctx.getBean(StandardConfig.class); - ctx.getBean(StandardIntegrationFlow.class).stop(); assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + ctx.getBean(SourcePollingChannelAdapter.class).stop(); List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); assertThat(sfCalls).containsExactly(1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3); File f1 = new File(tmpDir + File.separator + "fair" + File.separator + "f1"); @@ -135,7 +169,7 @@ public class RotatingServersTests extends FtpTestSupport { AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(VariableLocalConfig.class); StandardConfig config = ctx.getBean(StandardConfig.class); assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); - ctx.getBean(StandardIntegrationFlow.class).stop(); + ctx.getBean(SourcePollingChannelAdapter.class).stop(); List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1); File f1 = new File(tmpDir + File.separator + "variable" + File.separator + "foo" + File.separator + "f1"); @@ -152,15 +186,59 @@ public class RotatingServersTests extends FtpTestSupport { public void testStreaming() throws Exception { AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StreamingConfig.class); StandardConfig config = ctx.getBean(StandardConfig.class); - ctx.getBean(StandardIntegrationFlow.class).stop(); assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + ctx.getBean(SourcePollingChannelAdapter.class).stop(); List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); // there's an extra getSession() with this adapter in listFiles assertThat(sfCalls).containsExactly(1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 1, 1, 2, 2, 3); - assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3); + QueueChannel files = ctx.getBean("files", QueueChannel.class); + assertThat(files.getQueueSize()).isEqualTo(3); + Message received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f1"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f2"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f3"); ctx.close(); } + @Test + public void testFairStreaming() throws Exception { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(FairStreamingConfig.class); + try { + StandardConfig config = ctx.getBean(StandardConfig.class); + assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + ctx.getBean(SourcePollingChannelAdapter.class).stop(); + List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); + assertThat(sfCalls).containsExactly(1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3, 3, 1, 2, 3, 1, 2); + QueueChannel files = ctx.getBean("files", QueueChannel.class); + assertThat(files.getQueueSize()).isEqualTo(6); + Message received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f1"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f2"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f3"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f4"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f5"); + received = files.receive(0); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE, String.class)).isEqualTo("f6"); + } + finally { + ctx.close(); + } + } @Configuration @EnableIntegration @@ -307,4 +385,36 @@ public class RotatingServersTests extends FtpTestSupport { } + @Configuration + public static class FairStreamingConfig extends StandardConfig { + + @Override + @Bean + public RotatingServerAdvice advice() { + List keyDirectories = new ArrayList<>(); + keyDirectories.add(new RotationPolicy.KeyDirectory("one", "foo")); + keyDirectories.add(new RotationPolicy.KeyDirectory("two", "baz")); + keyDirectories.add(new RotationPolicy.KeyDirectory("three", "fiz")); + return theAdvice(keyDirectories); + } + + @Override + @Bean + public IntegrationFlow flow() { + return IntegrationFlows.from(Ftp.inboundStreamingAdapter(new FtpRemoteFileTemplate(sf())) + .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate")) + .remoteDirectory(".") + .maxFetchSize(1), + e -> e.poller(Pollers.fixedDelay(1).advice(advice()))) + .channel(MessageChannels.queue("files")) + .get(); + } + + @Override + protected RotatingServerAdvice theAdvice(List keyDirectories) { + return new RotatingServerAdvice(sf(), keyDirectories, true); + } + + } + }