From a167290f9bcc965d2583956bd04bad64fba2b51a 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 --- ...tractRemoteFileStreamingMessageSource.java | 78 +++++++++++++---- .../file/remote/StreamingInboundTests.java | 87 +++++++++++++++---- .../FtpStreamingMessageSourceTests.java | 14 ++- .../jdbc/StoredProcJavaConfigTests.java | 3 +- .../SftpStreamingMessageSourceTests.java | 3 + 5 files changed, 150 insertions(+), 35 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 69475a3c1d..7d5826d886 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 @@ -28,10 +28,13 @@ 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.Test; @@ -40,13 +43,16 @@ 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 @@ -67,6 +73,7 @@ public class StreamingInboundTests { streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/foo"); streamer.afterPropertiesSet(); + streamer.start(); Message received = (Message) this.transformer.transform(streamer.receive()); assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes()); assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); @@ -112,6 +119,7 @@ public class StreamingInboundTests { streamer.setMaxFetchSize(1); streamer.setFilter(new AcceptOnceFileListFilter<>()); streamer.afterPropertiesSet(); + streamer.start(); Message received = (Message) this.transformer.transform(streamer.receive()); assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes()); assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); @@ -138,17 +146,18 @@ public class StreamingInboundTests { streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/bad"); streamer.afterPropertiesSet(); - assertThatExceptionOfType(MessagingException.class) + 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)); @@ -160,7 +169,7 @@ public class StreamingInboundTests { assertThat(received.getPayload()).isEqualTo("foo"); assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); - received = (Message) out.receive(0); + 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"); @@ -185,10 +194,43 @@ public class StreamingInboundTests { verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(5)).close(); } + @SuppressWarnings("unchecked") + @Test + public void testStopAdapterRemovesUnprocessed() { + 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() { + 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 @@ -252,11 +294,7 @@ public class StreamingInboundTests { @Override public String getFileInfo() { - return asString(); - } - - private String asString() { - return "StringFileInfo [name=" + this.name + "]"; + return name; } } @@ -271,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); @@ -293,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; } @@ -309,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 2ff5e9e4af..c4b76b8e1c 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 @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; 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,6 +49,8 @@ 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; @@ -114,12 +117,17 @@ 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(); - assertThat(this.data.receive(10000)).isNotNull(); received = (Message) this.data.receive(10000); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO)).isInstanceOf(FtpFileInfo.class); + assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).hasSize(1); + assertThat(this.metadataMap).hasSize(2); this.adapter.stop(); + assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).isEmpty(); + assertThat(this.metadataMap).hasSize(1); } @Test @@ -127,6 +135,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { FtpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(new AcceptAllFileListFilter<>()); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo(" ftpSource1.txt"); @@ -142,6 +151,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { FtpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(null); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo(" ftpSource1.txt"); @@ -174,7 +184,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { public PollerMetadata defaultPoller() { PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setTrigger(new PeriodicTrigger(500)); - pollerMetadata.setMaxMessagesPerPoll(1); + 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 6f692b9c20..87f07de288 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 @@ -77,7 +77,8 @@ public class StoredProcJavaConfigTests { public void test() { Message received = fooChannel.receive(10000); assertThat(received).isNotNull(); - Collection primes = (Collection) received.getPayload(); + @SuppressWarnings("unchecked") + Collection primes = (Collection) received.getPayload(); assertThat(primes).containsExactly(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 22803a5af7..e9bdca6009 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 @@ -127,6 +127,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { SftpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(new AcceptAllFileListFilter<>()); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)) @@ -140,6 +141,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { SftpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(null); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)) @@ -153,6 +155,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { SftpStreamingMessageSource messageSource = buildSource(); messageSource.setFilter(Arrays::asList); messageSource.afterPropertiesSet(); + messageSource.start(); Message received = messageSource.receive(); assertThat(received).isNotNull(); assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE))