GH-2776: Fix Streaming Remote File MessageSource

Fixes #2776
Also see #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

Author: garyrussell

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 @@
 
 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.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,15 +51,17 @@
  * @since 4.3
  *
  */
-public abstract class AbstractRemoteFileStreamingMessageSource<F> extends AbstractMessageSource<InputStream>
-		implements BeanFactoryAware, InitializingBean {
+public abstract class AbstractRemoteFileStreamingMessageSource<F>
+		extends AbstractMessageSource<InputStream> implements Lifecycle {
 
 	private final RemoteFileTemplate<F> remoteFileTemplate;
 
 	private final BlockingQueue<AbstractFileInfo<F>> toBeReceived = new LinkedBlockingQueue<AbstractFileInfo<F>>();
 
 	private final Comparator<AbstractFileInfo<F>> comparator;
 
+	private final AtomicBoolean running = new AtomicBoolean();
+
 	/**
 	 * the path on the remote server.
 	 */
@@ -107,7 +113,11 @@ public void setRemoteFileSeparator(String remoteFileSeparator) {
 	 * @param filter the file list filter.
 	 */
 	public void setFilter(FileListFilter<F> filter) {
-		this.filter = filter;
+		doSetFilter(filter);
+	}
+
+	protected final void doSetFilter(FileListFilter<F> filterToSet) {
+		this.filter = filterToSet;
 	}
 
 	protected RemoteFileTemplate<F> getRemoteFileTemplate() {
@@ -127,26 +137,67 @@ public final void afterPropertiesSet() {
 	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<F> 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<F> 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<F> 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<F>) this.filter).remove(file.getFileInfo());
+		}
+	}
+
 	protected AbstractFileInfo<F> poll() {
 		if (this.toBeReceived.size() == 0) {
 			listFiles();

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 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.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 void testAllData() throws Exception {
 		streamer.setBeanFactory(mock(BeanFactory.class));
 		streamer.setRemoteDirectory("/foo");
 		streamer.afterPropertiesSet();
+		streamer.start();
 		Message<byte[]> received = (Message<byte[]>) 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 void testExceptionOnFetch() {
 		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 void testLineByLine() throws Exception {
 
 	public static class Streamer extends AbstractRemoteFileStreamingMessageSource<String> {
 
+		ConcurrentHashMap<String, String> metadataMap = new ConcurrentHashMap<>();
+
 		protected Streamer(RemoteFileTemplate<String> template, Comparator<AbstractFileInfo<String>> comparator) {
 			super(template, comparator);
+			doSetFilter(new StringPersistentFileListFilter(new SimpleMetadataStore(this.metadataMap), "streamer"));
 		}
 
 		@Override
@@ -151,7 +166,7 @@ public String getComponentType() {
 
 		@Override
 		protected List<AbstractFileInfo<String>> asFileInfoList(Collection<String> files) {
-			List<AbstractFileInfo<String>> infos = new ArrayList<AbstractFileInfo<String>>();
+			List<AbstractFileInfo<String>> infos = new ArrayList<>();
 			for (String file : files) {
 				infos.add(new StringFileInfo(file));
 			}
@@ -200,7 +215,7 @@ public String getPermissions() {
 
 		@Override
 		public String getFileInfo() {
-			return null;
+			return name;
 		}
 
 	}
@@ -215,9 +230,14 @@ public StringRemoteFileTemplate(SessionFactory<String> sessionFactory) {
 
 	public static class StringSessionFactory implements SessionFactory<String> {
 
+		private Session<String> singletonSession;
+
 		@SuppressWarnings("unchecked")
 		@Override
 		public Session<String> getSession() {
+			if (this.singletonSession != null) {
+				return this.singletonSession;
+			}
 			try {
 				Session<String> session = mock(Session.class);
 				willReturn(new String[] { "/foo/foo", "/foo/bar" }).given(session).list("/foo");
@@ -232,10 +252,14 @@ public Session<String> getSession() {
 				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 Session<String> getSession() {
 
 	}
 
+	public static class StringPersistentFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<String> {
+
+		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;
+		}
+
+	}
+
 }

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.