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

Author: garyrussell

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 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -26,19 +27,20 @@
 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.messaging.MessagingException;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
@@ -53,14 +55,16 @@
  *
  */
 public abstract class AbstractRemoteFileStreamingMessageSource<F>
-		extends AbstractFetchLimitingMessageSource<InputStream> implements BeanFactoryAware, InitializingBean {
+		extends AbstractFetchLimitingMessageSource<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();
+
 	private boolean fileInfoJson = true;
 
 	/**
@@ -117,8 +121,8 @@ public void setFilter(FileListFilter<F> filter) {
 		doSetFilter(filter);
 	}
 
-	protected final void doSetFilter(FileListFilter<F> filter) {
-		this.filter = filter;
+	protected final void doSetFilter(FileListFilter<F> filterToSet) {
+		this.filter = filterToSet;
 	}
 
 	/**
@@ -149,23 +153,53 @@ 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())
-						.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 @@ protected Object doReceive(int maxFetchSize) {
 		return doReceive();
 	}
 
+	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();