From 45789a99b850245ef0a79990f4fa51892d411b36 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 # Conflicts: # spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java # spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java # Conflicts: # spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java # 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-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java --- ...tractRemoteFileStreamingMessageSource.java | 79 +++++++++++++++---- .../file/remote/StreamingInboundTests.java | 56 +++++++++++-- .../FtpStreamingMessageSourceTests.java | 2 +- 3 files changed, 115 insertions(+), 22 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 632ffc7221..a9725f3bcc 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 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.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -26,15 +27,18 @@ import java.util.Iterator; 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.AbstractMessageSource; 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.messaging.MessagingException; import org.springframework.util.Assert; @@ -47,8 +51,8 @@ import org.springframework.util.Assert; * @since 4.3 * */ -public abstract class AbstractRemoteFileStreamingMessageSource extends AbstractMessageSource - implements BeanFactoryAware, InitializingBean { +public abstract class AbstractRemoteFileStreamingMessageSource + extends AbstractMessageSource implements Lifecycle { private final RemoteFileTemplate remoteFileTemplate; @@ -56,6 +60,8 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra private final Comparator> comparator; + private final AtomicBoolean running = new AtomicBoolean(); + /** * the path on the remote server. */ @@ -107,7 +113,11 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra * @param filter the file list filter. */ public void setFilter(FileListFilter filter) { - this.filter = filter; + doSetFilter(filter); + } + + protected final void doSetFilter(FileListFilter filterToSet) { + this.filter = filterToSet; } protected RemoteFileTemplate getRemoteFileTemplate() { @@ -127,26 +137,67 @@ public abstract class AbstractRemoteFileStreamingMessageSource extends Abstra 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()) - .build(); + 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()) + .build(); + } + catch (IOException e) { + throw new MessagingException("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; } + 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 4fc6c6f40b..f67e20f7d2 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 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. @@ -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.Rule; import org.junit.Test; @@ -41,15 +44,22 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; 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 * */ @@ -67,6 +77,7 @@ 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)); @@ -90,16 +101,17 @@ public class StreamingInboundTests { streamer.setBeanFactory(mock(BeanFactory.class)); streamer.setRemoteDirectory("/bad"); streamer.afterPropertiesSet(); - streamer.receive(); + streamer.start(); } - @SuppressWarnings("unchecked") @Test + @SuppressWarnings("unchecked") 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)); @@ -140,8 +152,11 @@ public class StreamingInboundTests { 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 @@ -151,7 +166,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)); } @@ -200,7 +215,7 @@ public class StreamingInboundTests { @Override public String getFileInfo() { - return null; + return name; } } @@ -215,9 +230,14 @@ public class StreamingInboundTests { public static class StringSessionFactory implements SessionFactory { + private Session singletonSession; + @SuppressWarnings("unchecked") @Override public Session getSession() { + if (this.singletonSession != null) { + return this.singletonSession; + } try { Session session = mock(Session.class); willReturn(new String[] { "/foo/foo", "/foo/bar" }).given(session).list("/foo"); @@ -232,10 +252,14 @@ 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.singletonSession = session; + return session; } catch (Exception e) { @@ -245,4 +269,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 2aa678d5a2..63183e79eb 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-2017 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.