From 648760bb43ac3673b543404ecea6c87fd4fc76a8 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 1 Mar 2019 14:35:31 -0500 Subject: [PATCH] GH-2776: Fix Streaming Remote File MessageSource Fixes https://github.com/spring-projects/spring-integration/issues/2776 Also see https://github.com/spring-projects/spring-integration/issues/2777 - reset the filter for the current file if the fetch fails - implement `Lifecycle` and clear the `toBeReceived` queue and corresponding filter entries * Polishing - PR Comments **cherry-pick to all supported** * Polishing # Conflicts: # spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java # spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java # spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java --- ...tractRemoteFileStreamingMessageSource.java | 78 +++++-- .../file/remote/StreamingInboundTests.java | 192 +++++++++++------- .../FtpStreamingMessageSourceTests.java | 23 ++- .../jdbc/StoredProcJavaConfigTests.java | 3 +- .../SftpStreamingMessageSourceTests.java | 3 + 5 files changed, 204 insertions(+), 95 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 b4b6757350..814262f3f2 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-2018 the original author or authors. + * Copyright 2016-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. @@ -18,6 +18,7 @@ package org.springframework.integration.file.remote; import java.io.IOException; import java.io.InputStream; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -25,20 +26,21 @@ import java.util.Comparator; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; -import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.ResettableFileListFilter; 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.lang.Nullable; -import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -53,7 +55,7 @@ import org.springframework.util.ObjectUtils; * */ public abstract class AbstractRemoteFileStreamingMessageSource - extends AbstractFetchLimitingMessageSource implements BeanFactoryAware, InitializingBean { + extends AbstractFetchLimitingMessageSource implements Lifecycle { private final RemoteFileTemplate remoteFileTemplate; @@ -61,6 +63,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource private final Comparator comparator; + private final AtomicBoolean running = new AtomicBoolean(); + private boolean fileInfoJson = true; /** @@ -117,8 +121,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource doSetFilter(filter); } - protected final void doSetFilter(FileListFilter filter) { - this.filter = filter; + protected final void doSetFilter(FileListFilter filterToSet) { + this.filter = filterToSet; } /** @@ -149,23 +153,53 @@ public abstract class AbstractRemoteFileStreamingMessageSource protected void doInit() { } + + @Override + public void start() { + this.running.set(true); + } + + @Override + public void stop() { + if (this.running.compareAndSet(true, false)) { + // remove unprocessed files from the queue (and filter) + AbstractFileInfo file = this.toBeReceived.poll(); + while (file != null) { + resetFilterIfNecessary(file); + file = this.toBeReceived.poll(); + } + } + } + + @Override + public boolean isRunning() { + return this.running.get(); + } + @Override protected Object doReceive() { + Assert.state(this.running.get(), () -> getComponentName() + " is not running"); AbstractFileInfo file = poll(); if (file != null) { - 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_FILE_INFO, - this.fileInfoJson ? file.toJson() : file); + 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_FILE_INFO, + this.fileInfoJson ? file.toJson() : file); + } + catch (IOException e) { + throw new UncheckedIOException("IOException when retrieving " + remotePath, e); + } } - catch (IOException e) { - throw new MessagingException("IOException when retrieving " + remotePath, e); + catch (RuntimeException e) { + resetFilterIfNecessary(file); + throw e; } } return null; @@ -176,6 +210,16 @@ public abstract class AbstractRemoteFileStreamingMessageSource return doReceive(); } + private void resetFilterIfNecessary(AbstractFileInfo file) { + if (this.filter instanceof ResettableFileListFilter) { + if (this.logger.isInfoEnabled()) { + this.logger.info("Removing the remote file '" + file + + "' from the filter for a subsequent transfer attempt"); + } + ((ResettableFileListFilter) this.filter).remove(file.getFileInfo()); + } + } + protected AbstractFileInfo poll() { if (this.toBeReceived.size() == 0) { listFiles(); 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 44c7013c8d..e3d2c3ee2e 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-2018 the original author or authors. + * Copyright 2016-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. @@ -16,10 +16,8 @@ package org.springframework.integration.file.remote; -import static org.hamcrest.CoreMatchers.containsString; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertThat; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willReturn; import static org.mockito.BDDMockito.willThrow; @@ -30,38 +28,41 @@ import static org.mockito.Mockito.verify; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collection; import java.util.Comparator; import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; -import org.junit.Rule; import org.junit.Test; -import org.junit.rules.ExpectedException; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.StreamTransformer; import org.springframework.messaging.Message; -import org.springframework.messaging.MessagingException; /** * @author Gary Russell + * @author Artem Bilan + * * @since 4.3 * */ public class StreamingInboundTests { - @Rule - public ExpectedException exception = ExpectedException.none(); - private final StreamTransformer transformer = new StreamTransformer(); @SuppressWarnings("unchecked") @@ -72,34 +73,35 @@ public class StreamingInboundTests { streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/foo"); streamer.afterPropertiesSet(); + streamer.start(); Message received = (Message) this.transformer.transform(streamer.receive()); - assertEquals("foo\nbar", new String(received.getPayload())); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes()); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); - assertThat(fileInfo, containsString("remoteDirectory\":\"/foo")); - assertThat(fileInfo, containsString("permissions\":\"-rw-rw-rw")); - assertThat(fileInfo, containsString("size\":42")); - assertThat(fileInfo, containsString("directory\":false")); - assertThat(fileInfo, containsString("filename\":\"foo")); - assertThat(fileInfo, containsString("modified\":42000")); - assertThat(fileInfo, containsString("link\":false")); + assertThat(fileInfo).contains("remoteDirectory\":\"/foo"); + assertThat(fileInfo).contains("permissions\":\"-rw-rw-rw"); + assertThat(fileInfo).contains("size\":42"); + assertThat(fileInfo).contains("directory\":false"); + assertThat(fileInfo).contains("filename\":\"foo"); + assertThat(fileInfo).contains("modified\":42000"); + assertThat(fileInfo).contains("link\":false"); // close after list, transform verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(2)).close(); received = (Message) this.transformer.transform(streamer.receive()); - assertEquals("baz\nqux", new String(received.getPayload())); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertThat(received.getPayload()).isEqualTo("baz\nqux".getBytes()); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); - assertThat(fileInfo, containsString("remoteDirectory\":\"/foo")); - assertThat(fileInfo, containsString("permissions\":\"-rw-rw-rw")); - assertThat(fileInfo, containsString("size\":42")); - assertThat(fileInfo, containsString("directory\":false")); - assertThat(fileInfo, containsString("filename\":\"bar")); - assertThat(fileInfo, containsString("modified\":42000")); - assertThat(fileInfo, containsString("link\":false")); + assertThat(fileInfo).contains("remoteDirectory\":\"/foo"); + assertThat(fileInfo).contains("permissions\":\"-rw-rw-rw"); + assertThat(fileInfo).contains("size\":42"); + assertThat(fileInfo).contains("directory\":false"); + assertThat(fileInfo).contains("filename\":\"bar"); + assertThat(fileInfo).contains("modified\":42000"); + assertThat(fileInfo).contains("link\":false"); // close after transform verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(3)).close(); @@ -117,18 +119,19 @@ public class StreamingInboundTests { streamer.setMaxFetchSize(1); streamer.setFilter(new AcceptOnceFileListFilter<>()); streamer.afterPropertiesSet(); + streamer.start(); Message received = (Message) this.transformer.transform(streamer.receive()); - assertEquals("foo\nbar", new String(received.getPayload())); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes()); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); // close after list, transform verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(2)).close(); received = (Message) this.transformer.transform(streamer.receive()); - assertEquals("baz\nqux", new String(received.getPayload())); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); + assertThat(received.getPayload()).isEqualTo("baz\nqux".getBytes()); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); // close after list, transform verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(4)).close(); @@ -138,22 +141,23 @@ public class StreamingInboundTests { @Test public void testExceptionOnFetch() throws Exception { - exception.expect(MessagingException.class); StringSessionFactory sessionFactory = new StringSessionFactory(); Streamer streamer = new Streamer(new StringRemoteFileTemplate(sessionFactory), null); streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/bad"); streamer.afterPropertiesSet(); - streamer.receive(); + streamer.start(); + assertThatExceptionOfType(UncheckedIOException.class) + .isThrownBy(streamer::receive); } - @SuppressWarnings("unchecked") @Test public void testLineByLine() throws Exception { Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null); streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/foo"); streamer.afterPropertiesSet(); + streamer.start(); QueueChannel out = new QueueChannel(); FileSplitter splitter = new FileSplitter(); splitter.setBeanFactory(mock(BeanFactory.class)); @@ -161,39 +165,72 @@ public class StreamingInboundTests { splitter.afterPropertiesSet(); Message receivedStream = streamer.receive(); splitter.handleMessage(receivedStream); - Message received = (Message) out.receive(0); - assertEquals("foo", received.getPayload()); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); - received = (Message) out.receive(0); - assertEquals("bar", received.getPayload()); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE)); - assertNull(out.receive(0)); + Message received = out.receive(0); + assertThat(received.getPayload()).isEqualTo("foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); + received = out.receive(0); + assertThat(received.getPayload()).isEqualTo("bar"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); + assertThat(out.receive(0)).isNull(); // close by list, splitter verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(3)).close(); receivedStream = streamer.receive(); splitter.handleMessage(receivedStream); - received = (Message) out.receive(0); - assertEquals("baz", received.getPayload()); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); - received = (Message) out.receive(0); - assertEquals("qux", received.getPayload()); - assertEquals("/foo", received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); - assertEquals("bar", received.getHeaders().get(FileHeaders.REMOTE_FILE)); - assertNull(out.receive(0)); + received = out.receive(0); + assertThat(received.getPayload()).isEqualTo("baz"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); + received = out.receive(0); + assertThat(received.getPayload()).isEqualTo("qux"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); + assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); + assertThat(out.receive(0)).isNull(); // close by splitter verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(5)).close(); } + @SuppressWarnings("unchecked") + @Test + public void testStopAdapterRemovesUnprocessed() throws Exception { + Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null); + streamer.setBeanFactory(mock(BeanFactory.class)); + streamer.setRemoteDirectory("/foo"); + streamer.afterPropertiesSet(); + streamer.start(); + assertThat(streamer.receive()).isNotNull(); + assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1); + assertThat(streamer.metadataMap).hasSize(2); + streamer.stop(); + assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(0); + assertThat(streamer.metadataMap).hasSize(1); + } + + @SuppressWarnings("unchecked") + @Test + public void testFilterReversedOnBadFetch() throws Exception { + Streamer streamer = new Streamer(new StringRemoteFileTemplate(new StringSessionFactory()), null); + streamer.setBeanFactory(mock(BeanFactory.class)); + streamer.setRemoteDirectory("/bad"); + streamer.afterPropertiesSet(); + streamer.start(); + assertThatExceptionOfType(UncheckedIOException.class) + .isThrownBy(streamer::receive); + assertThat(TestUtils.getPropertyValue(streamer, "toBeReceived", BlockingQueue.class)).hasSize(1); + assertThat(streamer.metadataMap).hasSize(1); + } + public static class Streamer extends AbstractRemoteFileStreamingMessageSource { + ConcurrentHashMap metadataMap = new ConcurrentHashMap<>(); + protected Streamer(RemoteFileTemplate template, Comparator comparator) { super(template, comparator); + doSetFilter(new StringPersistentFileListFilter(new SimpleMetadataStore(this.metadataMap), "streamer")); } @Override @@ -203,7 +240,7 @@ public class StreamingInboundTests { @Override protected List> asFileInfoList(Collection files) { - List> infos = new ArrayList>(); + List> infos = new ArrayList<>(); for (String file : files) { infos.add(new StringFileInfo(file)); } @@ -257,11 +294,7 @@ public class StreamingInboundTests { @Override public String getFileInfo() { - return asString(); - } - - private String asString() { - return "StringFileInfo [name=" + this.name + "]"; + return name; } } @@ -276,13 +309,13 @@ public class StreamingInboundTests { public static class StringSessionFactory implements SessionFactory { - private Session session; + private Session singletonSession; @SuppressWarnings("unchecked") @Override public Session getSession() { - if (this.session != null) { - return this.session; + if (this.singletonSession != null) { + return this.singletonSession; } try { Session session = mock(Session.class); @@ -298,12 +331,13 @@ public class StreamingInboundTests { willReturn(foo2).given(session).readRaw("/bar/foo"); willReturn(bar2).given(session).readRaw("/bar/bar"); - willReturn(new String[] { "/bad/file" }).given(session).list("/bad"); - willThrow(new IOException("No file")).given(session).readRaw("/bad/file"); + willReturn(new String[] { "/bad/file1", "/bad/file2" }).given(session).list("/bad"); + willThrow(new IOException("No file")).given(session).readRaw("/bad/file1"); + willThrow(new IOException("No file")).given(session).readRaw("/bad/file2"); given(session.finalizeRaw()).willReturn(true); - this.session = session; + this.singletonSession = session; return session; } @@ -314,4 +348,22 @@ public class StreamingInboundTests { } + public static class StringPersistentFileListFilter extends AbstractPersistentAcceptOnceFileListFilter { + + public StringPersistentFileListFilter(ConcurrentMetadataStore store, String prefix) { + super(store, prefix); + } + + @Override + protected long modified(String file) { + return 0; + } + + @Override + protected String fileName(String file) { + return file; + } + + } + } 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 00783c068a..16a2ab593c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-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. @@ -17,14 +17,16 @@ package org.springframework.integration.ftp.inbound; import static org.hamcrest.CoreMatchers.containsString; -import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.io.Closeable; import java.io.InputStream; import java.util.Comparator; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -48,10 +50,11 @@ import org.springframework.integration.file.filters.AcceptAllFileListFilter; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.FtpTestSupport; import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; -import org.springframework.integration.ftp.session.FtpFileInfo; import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.test.util.OnlyOnceTrigger; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.StreamTransformer; import org.springframework.messaging.Message; import org.springframework.scheduling.support.PeriodicTrigger; @@ -118,12 +121,16 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { this.source.setFileInfoJson(false); this.data.purge(null); this.metadataMap.clear(); + this.adapter.setTrigger(new OnlyOnceTrigger()); + this.adapter.setMaxMessagesPerPoll(1); this.adapter.start(); received = (Message) this.data.receive(10000); assertNotNull(received); + assertEquals(1, TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class).size()); + assertEquals(2, this.metadataMap.size()); this.adapter.stop(); - - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO), instanceOf(FtpFileInfo.class)); + assertTrue(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class).isEmpty()); + assertEquals(1, this.metadataMap.size()); } @Test @@ -131,6 +138,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { FtpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(new AcceptAllFileListFilter<>()); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertNotNull(received); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" ftpSource1.txt")); @@ -146,6 +154,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { FtpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(null); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertNotNull(received); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), equalTo(" ftpSource1.txt")); @@ -178,7 +187,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { public PollerMetadata defaultPoller() { PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setTrigger(new PeriodicTrigger(500)); - pollerMetadata.setMaxMessagesPerPoll(2000); + pollerMetadata.setMaxMessagesPerPoll(2); return pollerMetadata; } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java index 404c8f8224..c37b4dc07f 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java @@ -81,7 +81,8 @@ public class StoredProcJavaConfigTests { public void test() { Message received = fooChannel.receive(10000); assertNotNull(received); - Collection primes = (Collection) received.getPayload(); + @SuppressWarnings("unchecked") + Collection primes = (Collection) received.getPayload(); assertThat(primes, Matchers.contains(2, 3, 5, 7)); received = fooChannel.receive(100); // verify maxMessagesPerPoll == 1 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 b5e5a8e908..b7bff72988 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 @@ -123,6 +123,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { SftpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(new AcceptAllFileListFilter<>()); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertNotNull(received); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), @@ -136,6 +137,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { SftpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(null); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertNotNull(received); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE), @@ -149,6 +151,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { SftpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(Arrays::asList); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertNotNull(received); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE),