GH-3043: Add FileHeaders.REMOTE_HOST header (#3044)

* GH-3043: Add FileHeaders.REMOTE_HOST header

Fixes https://github.com/spring-projects/spring-integration/issues/3043

* Populate a `FileHeaders.REMOTE_HOST` from the
`AbstractRemoteFileStreamingMessageSource` and "get"-based commands
in the `AbstractRemoteFileOutboundGateway`
* Extract the value from the a `Session.getHost()` contract
* The `AbstractInboundFileSynchronizingMessageSource` cannot be
addressed with this because the real message is already based on the
locally stored file
* Adjust some affected tests according our code style requirements

* * Add remote file info support into `AbstractInboundFileSynchronizingMessageSource`
* Introduce a `MetadataStore` functionality into the `AbstractInboundFileSynchronizer`
to gather a remote file info an save it in the URI style against local file
* Retrieve such an info in the `AbstractInboundFileSynchronizingMessageSource`
during local file polling
* Introduce `protocol()` contract for the `AbstractInboundFileSynchronizer`
to build a proper URI in the metadata for external readers to distinguish
remote files properly
* Document the feature

* * Fix some typos in Docs

* * Rename property and header constant to the `HOST_PORT` pair
* Fix typos in Docs
* Add  `remote-file-metadata-store` and `metadata-store-prefix` into XSD
of (S)FTP Inbound Channel Adapters
* Add `remoteFileMetadataStore` and `metadataStorePrefix` options
into `RemoteFileInboundChannelAdapterSpec` for Java DSL

spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java

@@ -52,4 +52,9 @@ public abstract class FileHeaders {
 	 */
 	public static final String REMOTE_FILE_INFO = PREFIX + "remoteFileInfo";
 
+	/**
+	 * A remote host/port the file has been polled from
+	 */
+	public static final String REMOTE_HOST_PORT = PREFIX + "remoteHostPort";
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java

@@ -45,14 +45,15 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 
 	@Override
 	protected final BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
-				this.getInboundFileSynchronizerClass());
+		BeanDefinitionBuilder synchronizerBuilder =
+				BeanDefinitionBuilder.genericBeanDefinition(getInboundFileSynchronizerClass());
 
 		synchronizerBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
 
 		// configure the InboundFileSynchronizer properties
-		BeanDefinition expressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression(
-				"remote-directory", "remote-directory-expression", parserContext, element, false);
+		BeanDefinition expressionDef =
+				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression(
+						"remote-directory", "remote-directory-expression", parserContext, element, false);
 		if (expressionDef != null) {
 			synchronizerBuilder.addPropertyValue("remoteDirectoryExpression", expressionDef);
 		}
@@ -62,6 +63,9 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		String remoteFileSeparator = element.getAttribute("remote-file-separator");
 		synchronizerBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "temporary-file-suffix");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(synchronizerBuilder, element,
+				"remote-file-metadata-store");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "metadata-store-prefix");
 
 		FileParserUtils.configureFilter(synchronizerBuilder, element, parserContext,
 				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass(),
@@ -100,6 +104,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 
 	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();
 
-	protected abstract Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass();
+	protected abstract Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>>
+	getPersistentAcceptOnceFileListFilterClass();
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java

@@ -29,6 +29,7 @@ import org.springframework.integration.file.filters.ExpressionFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
+import org.springframework.integration.metadata.MetadataStore;
 
 /**
  * A {@link MessageSourceSpec} for an {@link AbstractInboundFileSynchronizingMessageSource}.
@@ -245,15 +246,37 @@ public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFil
 		return _this();
 	}
 
+	/**
+	 * Configure a {@link MetadataStore} for remote files metadata.
+	 * @param remoteFileMetadataStore the {@link MetadataStore} to use.
+	 * @return the spec.
+	 * @since 5.2
+	 * @see AbstractInboundFileSynchronizer#setRemoteFileMetadataStore(MetadataStore)
+	 */
+	public S remoteFileMetadataStore(MetadataStore remoteFileMetadataStore) {
+		this.synchronizer.setRemoteFileMetadataStore(remoteFileMetadataStore);
+		return _this();
+	}
+
+	/**
+	 * Configure a prefix for remote files metadata keys.
+	 * @param metadataStorePrefix the metadata key prefix to use.
+	 * @return the spec.
+	 * @since 5.2
+	 * @see #remoteFileMetadataStore
+	 */
+	public S metadataStorePrefix(String metadataStorePrefix) {
+		this.synchronizer.setMetadataStorePrefix(metadataStorePrefix);
+		return _this();
+	}
+
 	@Override
 	public Map<Object, String> getComponentsToRegister() {
 		Map<Object, String> componentsToRegister = new LinkedHashMap<>();
 		componentsToRegister.put(this.synchronizer, null);
-
 		if (this.expressionFileListFilter != null) {
 			componentsToRegister.put(this.expressionFileListFilter, null);
 		}
-
 		return componentsToRegister;
 	}
 

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java

@@ -38,18 +38,16 @@ import org.springframework.util.Assert;
  *
  */
 public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F>
-		implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>,  Closeable {
+		implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, Closeable {
 
 	protected final ConcurrentMetadataStore store; // NOSONAR
 
+	protected final String prefix; // NOSONAR
+
 	@Nullable
 	protected final Flushable flushableStore; // NOSONAR
 
-	protected final String prefix; // NOSONAR
-
-	protected volatile boolean flushOnUpdate; // NOSONAR
-
-	private final Object monitor = new Object();
+	protected boolean flushOnUpdate; // NOSONAR
 
 	public AbstractPersistentAcceptOnceFileListFilter(ConcurrentMetadataStore store, String prefix) {
 		Assert.notNull(store, "'store' cannot be null");
@@ -76,20 +74,18 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 	@Override
 	public boolean accept(F file) {
 		String key = buildKey(file);
-		synchronized (this.monitor) {
-			String newValue = value(file);
-			String oldValue = this.store.putIfAbsent(key, newValue);
-			if (oldValue == null) { // not in store
-				flushIfNeeded();
-				return fileStillExists(file);
-			}
-			// same value in store
-			if (!isEqual(file, oldValue) && this.store.replace(key, oldValue, newValue)) {
-				flushIfNeeded();
-				return fileStillExists(file);
-			}
-			return false;
+		String newValue = value(file);
+		String oldValue = this.store.putIfAbsent(key, newValue);
+		if (oldValue == null) { // not in store
+			flushIfNeeded();
+			return fileStillExists(file);
 		}
+		// same value in store
+		if (!isEqual(file, oldValue) && this.store.replace(key, oldValue, newValue)) {
+			flushIfNeeded();
+			return fileStillExists(file);
+		}
+		return false;
 	}
 
 	/**
@@ -151,7 +147,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 	 * @return true if equal.
 	 */
 	protected boolean isEqual(F file, String value) {
-		return Long.valueOf(value) == modified(file);
+		return Long.parseLong(value) == modified(file);
 	}
 
 	/**

spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

@@ -205,6 +205,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 								.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
 								.setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
 								.setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
+								.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort())
 								.setHeader(FileHeaders.REMOTE_FILE_INFO,
 										this.fileInfoJson ? file.toJson() : file);
 					}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -72,7 +72,6 @@ import org.springframework.util.StringUtils;
  */
 public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReplyProducingMessageHandler {
 
-
 	private final RemoteFileTemplate<F> remoteFileTemplate;
 
 	private final Command command;
@@ -118,7 +117,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory,
 			MessageSessionCallback<F, ?> messageSessionCallback) {
 
-		this(new RemoteFileTemplate<F>(sessionFactory), messageSessionCallback);
+		this(new RemoteFileTemplate<>(sessionFactory), messageSessionCallback);
 	}
 
 	/**
@@ -161,7 +160,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, Command command,
 			@Nullable String expression) {
 
-		this(new RemoteFileTemplate<F>(sessionFactory), command, expression);
+		this(new RemoteFileTemplate<>(sessionFactory), command, expression);
 	}
 
 	/**
@@ -317,7 +316,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	 * @since 4.3
 	 */
 	public void setRenameExpression(Expression renameExpression) {
-		this.renameProcessor = new ExpressionEvaluatingMessageProcessor<String>(renameExpression);
+		this.renameProcessor = new ExpressionEvaluatingMessageProcessor<>(renameExpression);
 	}
 
 	/**
@@ -490,10 +489,14 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			dir += this.remoteFileTemplate.getRemoteFileSeparator();
 		}
 		final String fullDir = dir;
-		List<?> payload = this.remoteFileTemplate.execute(session -> ls(requestMessage, session, fullDir));
-		return getMessageBuilderFactory()
-				.withPayload(payload)
-				.setHeader(FileHeaders.REMOTE_DIRECTORY, dir);
+		return this.remoteFileTemplate.execute(session -> {
+			List<?> payload = ls(requestMessage, session, fullDir);
+			return getMessageBuilderFactory()
+					.withPayload(payload)
+					.setHeader(FileHeaders.REMOTE_DIRECTORY, fullDir)
+					.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
+		});
+
 	}
 
 	private Object doNlst(Message<?> requestMessage) {
@@ -504,11 +507,13 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			dir += this.remoteFileTemplate.getRemoteFileSeparator();
 		}
 		final String fullDir = dir;
-		List<?> payload = this.remoteFileTemplate.execute(session -> nlst(requestMessage, session, fullDir));
-
-		return getMessageBuilderFactory()
-				.withPayload(payload)
-				.setHeader(FileHeaders.REMOTE_DIRECTORY, dir);
+		return this.remoteFileTemplate.execute(session -> {
+			List<?> payload = nlst(requestMessage, session, fullDir);
+			return getMessageBuilderFactory()
+					.withPayload(payload)
+					.setHeader(FileHeaders.REMOTE_DIRECTORY, fullDir)
+					.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
+		});
 	}
 
 	/**
@@ -541,6 +546,12 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			session = this.remoteFileTemplate.getSessionFactory().getSession();
 			try {
 				payload = session.readRaw(remoteFilePath);
+				return getMessageBuilderFactory()
+						.withPayload(payload)
+						.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
+						.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
+						.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort())
+						.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session);
 			}
 			catch (IOException e) {
 				throw new MessageHandlingException(requestMessage,
@@ -549,26 +560,30 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			}
 		}
 		else {
-			payload = this.remoteFileTemplate.execute(session1 ->
-					get(requestMessage, session1, remoteDir, remoteFilePath, remoteFilename, null));
+			return this.remoteFileTemplate.execute(session1 -> {
+				Object getPayload = get(requestMessage, session1, remoteDir, remoteFilePath, remoteFilename, null);
+				return getMessageBuilderFactory()
+						.withPayload(getPayload)
+						.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
+						.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
+						.setHeader(FileHeaders.REMOTE_HOST_PORT, session1.getHostPort());
+			});
 		}
-		return getMessageBuilderFactory()
-				.withPayload(payload)
-				.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
-				.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
-				.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session);
 	}
 
 	private Object doMget(final Message<?> requestMessage) {
 		String remoteFilePath = obtainRemoteFilePath(requestMessage);
-		final String remoteFilename = getRemoteFilename(remoteFilePath);
-		final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
-		List<File> payload = this.remoteFileTemplate.execute(session ->
-				mGet(requestMessage, session, remoteDir, remoteFilename));
-		return getMessageBuilderFactory()
-				.withPayload(payload)
-				.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
-				.setHeader(FileHeaders.REMOTE_FILE, remoteFilename);
+		String remoteFilename = getRemoteFilename(remoteFilePath);
+		String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
+		return this.remoteFileTemplate.execute(session -> {
+					List<File> payload = mGet(requestMessage, session, remoteDir, remoteFilename);
+					return getMessageBuilderFactory()
+							.withPayload(payload)
+							.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
+							.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
+							.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
+				}
+		);
 	}
 
 	private Object doRm(Message<?> requestMessage) {
@@ -576,12 +591,14 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		String remoteFilename = getRemoteFilename(remoteFilePath);
 		String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
 
-		boolean payload = this.remoteFileTemplate.execute(session -> rm(requestMessage, session, remoteFilePath));
-
-		return getMessageBuilderFactory()
-				.withPayload(payload)
-				.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
-				.setHeader(FileHeaders.REMOTE_FILE, remoteFilename);
+		return this.remoteFileTemplate.execute(session -> {
+			boolean payload = rm(requestMessage, session, remoteFilePath);
+			return getMessageBuilderFactory()
+					.withPayload(payload)
+					.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
+					.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
+					.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
+		});
 	}
 
 	/**
@@ -606,15 +623,16 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		String remoteFileNewPath = this.renameProcessor.processMessage(requestMessage);
 		Assert.hasLength(remoteFileNewPath, "New filename cannot be empty");
 
-		Boolean result =
-				this.remoteFileTemplate.execute(session ->
-						mv(requestMessage, session, remoteFilePath, remoteFileNewPath));
-
-		return getMessageBuilderFactory()
-				.withPayload(result)
-				.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
-				.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
-				.setHeader(FileHeaders.RENAME_TO, remoteFileNewPath);
+		return this.remoteFileTemplate.execute(session -> {
+					Boolean result = mv(requestMessage, session, remoteFilePath, remoteFileNewPath);
+					return getMessageBuilderFactory()
+							.withPayload(result)
+							.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
+							.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
+							.setHeader(FileHeaders.RENAME_TO, remoteFileNewPath)
+							.setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort());
+				}
+		);
 	}
 
 	private String obtainRemoteFilePath(Message<?> requestMessage) {

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java

@@ -37,11 +37,13 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Alen Turkovic
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBean {
 
-	private static final Log logger = LogFactory.getLog(CachingSessionFactory.class);
+	private static final Log LOGGER = LogFactory.getLog(CachingSessionFactory.class);
 
 	private final SessionFactory<F> sessionFactory;
 
@@ -55,7 +57,6 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 
 	/**
 	 * Create a CachingSessionFactory with an unlimited number of sessions.
-	 *
 	 * @param sessionFactory the underlying session factory.
 	 */
 	public CachingSessionFactory(SessionFactory<F> sessionFactory) {
@@ -66,11 +67,9 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	 * Create a CachingSessionFactory with the specified session limit. By default, if
 	 * no sessions are available in the cache, and the size limit has been reached,
 	 * calling threads will block until a session is available.
-	 * <p>
-	 * Do not cache a {@link DelegatingSessionFactory}, cache each delegate therein instead.
+	 * <p> Do not cache a {@link DelegatingSessionFactory}, cache each delegate therein instead.
 	 * @see #setSessionWaitTimeout(long)
 	 * @see #setPoolSize(int)
-	 *
 	 * @param sessionFactory The underlying session factory.
 	 * @param sessionCacheSize The maximum cache size.
 	 */
@@ -78,7 +77,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 		Assert.isTrue(!(sessionFactory instanceof DelegatingSessionFactory),
 				"'sessionFactory' cannot be a 'DelegatingSessionFactory'; cache each delegate instead");
 		this.sessionFactory = sessionFactory;
-		this.pool = new SimplePool<Session<F>>(sessionCacheSize, new SimplePool.PoolItemCallback<Session<F>>() {
+		this.pool = new SimplePool<>(sessionCacheSize, new SimplePool.PoolItemCallback<Session<F>>() {
 			@Override
 			public Session<F> createForPool() {
 				return CachingSessionFactory.this.sessionFactory.getSession();
@@ -100,7 +99,6 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 
 	/**
 	 * Sets the limit of how long to wait for a session to become available.
-	 *
 	 * @param sessionWaitTimeout the session wait timeout.
 	 * @throws IllegalStateException if the wait expires prior to a Session becoming available.
 	 */
@@ -111,7 +109,6 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	/**
 	 * Modify the target session pool size; the actual pool size will adjust up/down
 	 * to this size as and when sessions are requested or retrieved.
-	 *
 	 * @param poolSize The pool size.
 	 */
 	public void setPoolSize(int poolSize) {
@@ -148,9 +145,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 	 * returned to the cache.
 	 */
 	public synchronized void resetCache() {
-		if (logger.isDebugEnabled()) {
-			logger.debug("Cache reset; idle sessions will be removed, in-use sessions will be closed when returned");
-		}
+		LOGGER.debug("Cache reset; idle sessions will be removed, in-use sessions will be closed when returned");
 		if (this.isSharedSessionCapable && ((SharedSessionCapable) this.sessionFactory).isSharedSession()) {
 			((SharedSessionCapable) this.sessionFactory).resetSharedSession();
 		}
@@ -169,7 +164,7 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 		this.pool.removeAllIdleItems();
 	}
 
-	public class CachedSession implements Session<F> { //NOSONAR (final)
+	public class CachedSession implements Session<F> { //NOSONAR must be final, but can't for mocking in tests
 
 		private final Session<F> targetSession;
 
@@ -190,17 +185,17 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 		@Override
 		public synchronized void close() {
 			if (this.released) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Session " + this.targetSession + " already released.");
+				if (LOGGER.isDebugEnabled()) {
+					LOGGER.debug("Session " + this.targetSession + " already released.");
 				}
 			}
 			else {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Releasing Session " + this.targetSession + " back to the pool.");
+				if (LOGGER.isDebugEnabled()) {
+					LOGGER.debug("Releasing Session " + this.targetSession + " back to the pool.");
 				}
 				if (this.sharedSessionEpoch != CachingSessionFactory.this.sharedSessionEpoch) {
-					if (logger.isDebugEnabled()) {
-						logger.debug("Closing session " + this.targetSession + " after reset.");
+					if (LOGGER.isDebugEnabled()) {
+						LOGGER.debug("Closing session " + this.targetSession + " after reset.");
 					}
 					this.targetSession.close();
 				}
@@ -295,6 +290,11 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 			return this.targetSession.getClientInstance();
 		}
 
+		@Override
+		public String getHostPort() {
+			return this.targetSession.getHostPort();
+		}
+
 	}
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java

@@ -108,6 +108,13 @@ public interface Session<F> extends Closeable {
 	 */
 	Object getClientInstance();
 
+	/**
+	 * Return the host:port pair this session is connected to.
+	 * @return the host:port pair this session is connected to.
+	 * @since 5.2
+	 */
+	String getHostPort();
+
 	/**
 	 * Test the session is still alive, e.g. when checking out from a pool.
 	 * The default implementation simply delegates to {@link #isOpen()}.

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java

@@ -22,6 +22,8 @@ import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
@@ -34,6 +36,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
@@ -48,10 +51,13 @@ import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileUtils;
+import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Base class charged with knowing how to connect to a remote file system,
@@ -72,7 +78,7 @@ import org.springframework.util.ObjectUtils;
  * @since 2.0
  */
 public abstract class AbstractInboundFileSynchronizer<F>
-		implements InboundFileSynchronizer, BeanFactoryAware, InitializingBean, Closeable {
+		implements InboundFileSynchronizer, BeanFactoryAware, BeanNameAware, InitializingBean, Closeable {
 
 	protected static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
 
@@ -99,7 +105,7 @@ public abstract class AbstractInboundFileSynchronizer<F>
 	/**
 	 * The current evaluation of the expression.
 	 */
-	private volatile String evaluatedRemoteDirectory;
+	private String evaluatedRemoteDirectory;
 
 	/**
 	 * An {@link FileListFilter} that runs against the <em>remote</em> file system view.
@@ -124,9 +130,14 @@ public abstract class AbstractInboundFileSynchronizer<F>
 	@Nullable
 	private Comparator<F> comparator;
 
+	private MetadataStore remoteFileMetadataStore = new SimpleMetadataStore();
+
+	private String metadataStorePrefix;
+
+	private String name;
+
 	/**
 	 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
-	 *
 	 * @param sessionFactory The session factory.
 	 */
 	public AbstractInboundFileSynchronizer(SessionFactory<F> sessionFactory) {
@@ -250,11 +261,37 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		this.preserveTimestamp = preserveTimestamp;
 	}
 
+	/**
+	 * Configure a {@link MetadataStore} to hold a remote file info (host, port, remote directory)
+	 * to transfer downstream in message headers when local file is pulled.
+	 * @param remoteFileMetadataStore the {@link MetadataStore} to use.
+	 * @since 5.2
+	 */
+	public void setRemoteFileMetadataStore(MetadataStore remoteFileMetadataStore) {
+		this.remoteFileMetadataStore = remoteFileMetadataStore;
+	}
+
+	/**
+	 * Specify a prefix for keys in metadata store do not clash with other keys in the shared store.
+	 * @param metadataStorePrefix the prefix to use.
+	 * @since 5.2
+	 * @see #setRemoteFileMetadataStore(MetadataStore)
+	 */
+	public void setMetadataStorePrefix(String metadataStorePrefix) {
+		this.metadataStorePrefix = metadataStorePrefix;
+	}
+
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
 	}
 
+	@Override
+	public void setBeanName(String name) {
+		this.name = name;
+	}
+
 	@Override
 	public final void afterPropertiesSet() {
 		Assert.state(this.remoteDirectoryExpression != null, "'remoteDirectoryExpression' must not be null");
@@ -262,6 +299,9 @@ public abstract class AbstractInboundFileSynchronizer<F>
 			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
 		}
 		evaluateRemoteDirectory();
+		if (!StringUtils.hasText(this.metadataStorePrefix)) {
+			this.metadataStorePrefix = this.name;
+		}
 		doInit();
 	}
 
@@ -334,7 +374,8 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 			for (F file : filteredFiles) {
 				if (filteringOneByOne) {
-					if ((maxFetchSize < 0 || accepted < maxFetchSize) && this.filter.accept(file)) { // NOSONAR never null
+					if ((maxFetchSize < 0 || accepted < maxFetchSize) && this.filter
+							.accept(file)) { // NOSONAR never null
 						accepted++;
 					}
 					else {
@@ -358,7 +399,7 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		try {
 			if (file != null && !copyFileToLocalDirectory(this.evaluatedRemoteDirectory, file,
 					localDirectory, session)) {
-				renamedFailed = false;
+				renamedFailed = true;
 			}
 		}
 		catch (RuntimeException | IOException e1) {
@@ -452,6 +493,17 @@ public abstract class AbstractInboundFileSynchronizer<F>
 				if (this.preserveTimestamp && !localFile.setLastModified(modified)) {
 					throw new IllegalStateException("Could not sent last modified on file: " + localFile);
 				}
+				String[] hostPort = session.getHostPort().split(":");
+				try {
+					String remoteFileMetadata =
+							new URI(protocol(), null, hostPort[0], Integer.parseInt(hostPort[1]),
+									'/' + remoteDirectoryPath, null, remoteFileName)
+									.toString();
+					this.remoteFileMetadataStore.put(buildMetadataKey(localFile), remoteFileMetadata);
+				}
+				catch (URISyntaxException ex) {
+					throw new IllegalStateException("Cannot create a remote file metadata", ex);
+				}
 				return true;
 			}
 			else {
@@ -528,10 +580,44 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		}
 	}
 
+	/**
+	 * Obtain a metadata for remote file associated with the provided local file.
+	 * @param localFile the local file to retrieve metadata for.
+	 * @return the metadata for remove file in the URI style:
+	 * {@code protocol://host:port/remoteDirectory#remoteFileName}
+	 * @since 5.2
+	 */
+	@Nullable
+	public String getRemoteFileMetadata(File localFile) {
+		String metadataKey = buildMetadataKey(localFile);
+		return this.remoteFileMetadataStore.get(metadataKey);
+	}
+
+	/**
+	 * Remove a metadata for remote file associated with the provided local file.
+	 * @param localFile the local file to remove metadata for.
+	 * @since 5.2
+	 */
+	public void removeRemoteFileMetadata(File localFile) {
+		String metadataKey = buildMetadataKey(localFile);
+		this.remoteFileMetadataStore.remove(metadataKey);
+	}
+
+	private String buildMetadataKey(File file) {
+		return this.metadataStorePrefix + file.getAbsolutePath();
+	}
+
 	protected abstract boolean isFile(F file);
 
 	protected abstract String getFilename(F file);
 
 	protected abstract long getModified(F file);
 
+	/**
+	 * Return the protocol this synchronizer works with.
+	 * @return the protocol this synchronizer works with.
+	 * @since 5.2
+	 */
+	protected abstract String protocol();
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java

@@ -19,6 +19,7 @@ package org.springframework.integration.file.remote.synchronizer;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.net.URI;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.regex.Pattern;
@@ -28,6 +29,7 @@ import org.springframework.context.Lifecycle;
 import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource;
 import org.springframework.integration.file.DefaultDirectoryScanner;
 import org.springframework.integration.file.DirectoryScanner;
+import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
@@ -95,7 +97,6 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 
 	private volatile boolean running;
 
-
 	public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer) {
 		this(synchronizer, null);
 	}
@@ -192,7 +193,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 			this.fileSource.setDirectory(this.localDirectory);
 			initFiltersAndScanner();
 			if (this.getBeanFactory() != null) {
-				this.fileSource.setBeanFactory(this.getBeanFactory());
+				this.fileSource.setBeanFactory(getBeanFactory());
 			}
 			this.fileSource.afterPropertiesSet();
 			this.synchronizer.afterPropertiesSet();
@@ -265,6 +266,16 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 			messageBuilder = this.fileSource.doReceive();
 		}
 
+		if (messageBuilder != null) {
+			String remoteFileUri = this.synchronizer.getRemoteFileMetadata(messageBuilder.getPayload());
+			if (remoteFileUri != null) {
+				URI uri = URI.create(remoteFileUri);
+				messageBuilder.setHeader(FileHeaders.REMOTE_HOST_PORT, uri.getHost() + ':' + uri.getPort())
+						.setHeader(FileHeaders.REMOTE_DIRECTORY, uri.getPath())
+						.setHeader(FileHeaders.REMOTE_FILE, uri.getFragment());
+			}
+		}
+
 		return messageBuilder;
 	}
 
@@ -274,7 +285,6 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 				Arrays.asList(this.localFileListFilter, new RegexPatternFileListFilter(completePattern)));
 	}
 
-
 	/**
 	 * The {@link FileReadingMessageSource} extension to increase visibility
 	 * for the {@link FileReadingMessageSource#doReceive()}

spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java

@@ -17,7 +17,8 @@
 package org.springframework.integration.file.remote.gateway;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doAnswer;
@@ -58,6 +59,7 @@ import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
@@ -81,11 +83,11 @@ public class RemoteFileOutboundGatewayTests {
 	public final TemporaryFolder tempFolder = new TemporaryFolder();
 
 
-	@Test(expected = IllegalArgumentException.class)
+	@Test
 	public void testBad() {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
-		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(sessionFactory, "bad", "payload");
-		gw.afterPropertiesSet();
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> new TestRemoteFileOutboundGateway(sessionFactory, "bad", "payload"));
 	}
 
 	@Test
@@ -93,13 +95,10 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(sessionFactory, "get", "payload");
 		gw.setFilter(new TestPatternFilter(""));
-		try {
-			gw.afterPropertiesSet();
-			fail("Exception expected");
-		}
-		catch (IllegalArgumentException e) {
-			assertThat(e.getMessage().startsWith("Filters are not supported")).isTrue();
-		}
+
+		assertThatIllegalArgumentException()
+				.isThrownBy(gw::afterPropertiesSet)
+				.withMessageStartingWith("Filters are not supported");
 	}
 
 	@Test
@@ -107,13 +106,10 @@ public class RemoteFileOutboundGatewayTests {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(sessionFactory, "rm", "payload");
 		gw.setFilter(new TestPatternFilter(""));
-		try {
-			gw.afterPropertiesSet();
-			fail("Exception expected");
-		}
-		catch (IllegalArgumentException e) {
-			assertThat(e.getMessage().startsWith("Filters are not supported")).isTrue();
-		}
+
+		assertThatIllegalArgumentException()
+				.isThrownBy(gw::afterPropertiesSet)
+				.withMessageStartingWith("Filters are not supported");
 	}
 
 	@Test
@@ -139,11 +135,6 @@ public class RemoteFileOutboundGatewayTests {
 		testMGetWildGuts("f1", "f2");
 	}
 
-	/**
-	 * Test a wildcard mget where the full path is returned for each file
-	 *
-	 * @throws Exception
-	 */
 	@Test
 	public void testMGetWildFullPath() {
 		testMGetWildGuts("testremote/f1", "testremote/f2");
@@ -547,7 +538,7 @@ public class RemoteFileOutboundGatewayTests {
 	}
 
 	@Test
-	public void testGet() throws Exception {
+	public void testGet() {
 		SessionFactory sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(sessionFactory, "get", "payload");
 		gw.setLocalDirectory(new File(this.tmpDir));
@@ -609,22 +600,16 @@ public class RemoteFileOutboundGatewayTests {
 
 		// default (null)
 		MessageBuilder<File> out;
-		try {
-			out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
-			fail("Exception expected");
-		}
-		catch (MessageHandlingException e) {
-			assertThat(e.getMessage()).contains("already exists");
-		}
+
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> gw.handleRequestMessage(new GenericMessage<>("f1")))
+				.withMessageContaining("already exists");
 
 		gw.setFileExistsMode(FileExistsMode.FAIL);
-		try {
-			out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
-			fail("Exception expected");
-		}
-		catch (MessageHandlingException e) {
-			assertThat(e.getMessage()).contains("already exists");
-		}
+
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> gw.handleRequestMessage(new GenericMessage<>("f1")))
+				.withMessageContaining("already exists");
 
 		gw.setFileExistsMode(FileExistsMode.IGNORE);
 		out = (MessageBuilder<File>) gw.handleRequestMessage(new GenericMessage<>("f1"));
@@ -652,7 +637,8 @@ public class RemoteFileOutboundGatewayTests {
 
 	@Test
 	public void testGetTempFileDelete() {
-		SessionFactory sessionFactory = mock(SessionFactory.class);
+		@SuppressWarnings("unchecked")
+		SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
 		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(sessionFactory, "get", "payload");
 		gw.setLocalDirectory(new File(this.tmpDir));
 		gw.afterPropertiesSet();
@@ -672,18 +658,15 @@ public class RemoteFileOutboundGatewayTests {
 			}
 
 		});
-		try {
-			gw.handleRequestMessage(new GenericMessage<>("f1"));
-			fail("Expected exception");
-		}
-		catch (MessagingException e) {
-			assertThat(e.getCause()).isInstanceOf(RuntimeException.class);
-			assertThat(e.getCause().getMessage()).isEqualTo("test remove .writing");
-			@SuppressWarnings("unchecked")
-			RemoteFileTemplate template = new RemoteFileTemplate(sessionFactory);
-			File outFile = new File(this.tmpDir + "/f1" + template.getTemporaryFileSuffix());
-			assertThat(outFile.exists()).isFalse();
-		}
+
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> gw.handleRequestMessage(new GenericMessage<>("f1")))
+				.withCauseInstanceOf(RuntimeException.class)
+				.withMessageContaining("test remove .writing");
+
+		RemoteFileTemplate<?> template = new RemoteFileTemplate<>(sessionFactory);
+		File outFile = new File(this.tmpDir + "/f1" + template.getTemporaryFileSuffix());
+		assertThat(outFile.exists()).isFalse();
 	}
 
 
@@ -743,8 +726,7 @@ public class RemoteFileOutboundGatewayTests {
 			}
 
 			@Override
-			public void read(String source, OutputStream outputStream)
-					throws IOException {
+			public void read(String source, OutputStream outputStream) throws IOException {
 				outputStream.write("testfile".getBytes());
 			}
 
@@ -844,13 +826,10 @@ public class RemoteFileOutboundGatewayTests {
 		verify(session).rename("foo/bar.txt.writing", "foo/bar.txt");
 
 		gw.setFileExistsMode(FileExistsMode.FAIL);
-		try {
-			path = (String) gw.handleRequestMessage(requestMessage);
-			fail("Exception expected");
-		}
-		catch (Exception e) {
-			assertThat(e.getMessage()).contains("The destination file already exists");
-		}
+
+		assertThatExceptionOfType(MessageDeliveryException.class)
+				.isThrownBy(() -> gw.handleRequestMessage(requestMessage))
+				.withMessageContaining("The destination file already exists");
 
 		gw.setFileExistsMode(FileExistsMode.REPLACE);
 		path = (String) gw.handleRequestMessage(requestMessage);
@@ -876,10 +855,9 @@ public class RemoteFileOutboundGatewayTests {
 	}
 
 	@Test
+	@SuppressWarnings("unchecked")
 	public void testMput() throws Exception {
-		@SuppressWarnings("unchecked")
 		SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
-		@SuppressWarnings("unchecked")
 		Session<TestLsEntry> session = mock(Session.class);
 		RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<>(sessionFactory);
 		template.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
@@ -897,7 +875,6 @@ public class RemoteFileOutboundGatewayTests {
 		tempFolder.newFile("qux.txt");
 		Message<File> requestMessage = MessageBuilder.withPayload(tempFolder.getRoot())
 				.build();
-		@SuppressWarnings("unchecked")
 		List<String> out = (List<String>) gw.handleRequestMessage(requestMessage);
 		assertThat(out.size()).isEqualTo(2);
 		assertThat(out.get(0)).isNotEqualTo(out.get(1));
@@ -906,10 +883,9 @@ public class RemoteFileOutboundGatewayTests {
 	}
 
 	@Test
+	@SuppressWarnings("unchecked")
 	public void testMputRecursive() throws Exception {
-		@SuppressWarnings("unchecked")
 		SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
-		@SuppressWarnings("unchecked")
 		Session<TestLsEntry> session = mock(Session.class);
 		RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<>(sessionFactory);
 		template.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
@@ -931,7 +907,6 @@ public class RemoteFileOutboundGatewayTests {
 
 		Message<File> requestMessage = MessageBuilder.withPayload(tempFolder.getRoot())
 				.build();
-		@SuppressWarnings("unchecked")
 		List<String> out = (List<String>) gw.handleRequestMessage(requestMessage);
 		assertThat(out.size()).isEqualTo(3);
 		assertThat(out.get(0)).isNotEqualTo(out.get(1));
@@ -941,10 +916,9 @@ public class RemoteFileOutboundGatewayTests {
 	}
 
 	@Test
+	@SuppressWarnings("unchecked")
 	public void testMputCollection() throws Exception {
-		@SuppressWarnings("unchecked")
 		SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
-		@SuppressWarnings("unchecked")
 		Session<TestLsEntry> session = mock(Session.class);
 		RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<>(sessionFactory);
 		template.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
@@ -963,7 +937,6 @@ public class RemoteFileOutboundGatewayTests {
 		files.add(tempFolder.newFile("buz.txt"));
 		Message<List<File>> requestMessage = MessageBuilder.withPayload(files)
 				.build();
-		@SuppressWarnings("unchecked")
 		List<String> out = (List<String>) gw.handleRequestMessage(requestMessage);
 		assertThat(out.size()).isEqualTo(2);
 		assertThat(out.get(0)).isNotEqualTo(out.get(1));
@@ -1022,7 +995,7 @@ public class RemoteFileOutboundGatewayTests {
 
 		@Override
 		public boolean isOpen() {
-			return open;
+			return this.open;
 		}
 
 		@Override
@@ -1050,6 +1023,11 @@ public class RemoteFileOutboundGatewayTests {
 			return null;
 		}
 
+		@Override
+		public String getHostPort() {
+			return null;
+		}
+
 	}
 
 

spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java

@@ -17,13 +17,12 @@
 package org.springframework.integration.file.remote.session;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.ByteArrayInputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 
@@ -31,13 +30,15 @@ import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.expression.common.LiteralExpression;
-import org.springframework.integration.file.remote.InputStreamCallback;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 3.0
  *
  */
@@ -46,7 +47,7 @@ public class CachingSessionFactoryTests {
 	@Test
 	public void testCacheAndReset() {
 		TestSessionFactory factory = new TestSessionFactory();
-		CachingSessionFactory<String> cache = new CachingSessionFactory<String>(factory);
+		CachingSessionFactory<String> cache = new CachingSessionFactory<>(factory);
 		cache.setTestSession(true);
 		Session<String> sess1 = cache.getSession();
 		assertThat(TestUtils.getPropertyValue(sess1, "targetSession.id")).isEqualTo("session:1");
@@ -83,25 +84,24 @@ public class CachingSessionFactoryTests {
 		when(factory.getSession()).thenReturn(session);
 		when(session.readRaw("foo")).thenReturn(new ByteArrayInputStream("".getBytes()));
 		when(session.finalizeRaw()).thenReturn(true);
-		CachingSessionFactory<Object> ccf = new CachingSessionFactory<Object>(factory);
-		RemoteFileTemplate<Object> template = new RemoteFileTemplate<Object>(ccf);
+		CachingSessionFactory<Object> ccf = new CachingSessionFactory<>(factory);
+		RemoteFileTemplate<Object> template = new RemoteFileTemplate<>(ccf);
 		template.setFileNameExpression(new LiteralExpression("foo"));
 		template.setBeanFactory(mock(BeanFactory.class));
 		template.afterPropertiesSet();
-		try {
-			template.get(new GenericMessage<String>("foo"), (InputStreamCallback) stream -> {
-				throw new RuntimeException("bar");
-			});
-			fail("Expected exception");
-		}
-		catch (Exception e) {
-			assertThat(e.getCause()).isInstanceOf(RuntimeException.class);
-			assertThat(e.getCause().getMessage()).isEqualTo("bar");
-		}
+
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() ->
+						template.get(new GenericMessage<>("foo"),
+								stream -> {
+									throw new RuntimeException("bar");
+								}))
+				.withCauseInstanceOf(RuntimeException.class)
+				.withMessageContaining("bar");
 		verify(session).close();
 	}
 
-	private class TestSessionFactory implements SessionFactory<String> {
+	private static class TestSessionFactory implements SessionFactory<String> {
 
 		private int n;
 
@@ -112,7 +112,7 @@ public class CachingSessionFactoryTests {
 
 	}
 
-	private class TestSession implements Session<String> {
+	private static class TestSession implements Session<String> {
 
 		@SuppressWarnings("unused")
 		private final String id;
@@ -127,39 +127,39 @@ public class CachingSessionFactoryTests {
 		}
 
 		@Override
-		public boolean remove(String path) throws IOException {
+		public boolean remove(String path) {
 			return false;
 		}
 
 		@Override
-		public String[] list(String path) throws IOException {
+		public String[] list(String path) {
 			return null;
 		}
 
 		@Override
-		public void read(String source, OutputStream outputStream) throws IOException {
+		public void read(String source, OutputStream outputStream) {
 		}
 
 		@Override
-		public void write(InputStream inputStream, String destination) throws IOException {
+		public void write(InputStream inputStream, String destination) {
 		}
 
 		@Override
-		public void append(InputStream inputStream, String destination) throws IOException {
+		public void append(InputStream inputStream, String destination) {
 		}
 
 		@Override
-		public boolean mkdir(String directory) throws IOException {
+		public boolean mkdir(String directory) {
 			return false;
 		}
 
 		@Override
-		public boolean rmdir(String directory) throws IOException {
+		public boolean rmdir(String directory) {
 			return false;
 		}
 
 		@Override
-		public void rename(String pathFrom, String pathTo) throws IOException {
+		public void rename(String pathFrom, String pathTo) {
 		}
 
 		@Override
@@ -173,22 +173,22 @@ public class CachingSessionFactoryTests {
 		}
 
 		@Override
-		public boolean exists(String path) throws IOException {
+		public boolean exists(String path) {
 			return false;
 		}
 
 		@Override
-		public String[] listNames(String path) throws IOException {
+		public String[] listNames(String path) {
 			return null;
 		}
 
 		@Override
-		public InputStream readRaw(String source) throws IOException {
+		public InputStream readRaw(String source) {
 			return null;
 		}
 
 		@Override
-		public boolean finalizeRaw() throws IOException {
+		public boolean finalizeRaw() {
 			return false;
 		}
 
@@ -197,6 +197,11 @@ public class CachingSessionFactoryTests {
 			return null;
 		}
 
+		@Override
+		public String getHostPort() {
+			return null;
+		}
+
 		@Override
 		public boolean test() {
 			this.testCalled = true;

spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java

@@ -17,7 +17,7 @@
 package org.springframework.integration.file.remote.synchronizer;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.Mockito.mock;
 
 import java.io.File;
@@ -72,6 +72,11 @@ public class AbstractRemoteFileSynchronizerTests {
 				return 0;
 			}
 
+			@Override
+			protected String protocol() {
+				return "file";
+			}
+
 			@Override
 			protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile,
 					File localDirectory, Session<String> session) throws IOException {
@@ -83,19 +88,14 @@ public class AbstractRemoteFileSynchronizerTests {
 			}
 
 		};
-		sync.setFilter(new AcceptOnceFileListFilter<String>());
+		sync.setFilter(new AcceptOnceFileListFilter<>());
 		sync.setRemoteDirectory("foo");
 
-		try {
-			sync.synchronizeToLocalDirectory(mock(File.class));
-			assertThat(count.get()).isEqualTo(1);
-			fail("Expected exception");
-		}
-		catch (MessagingException e) {
-			assertThat(e.getCause()).isInstanceOf(MessagingException.class);
-			assertThat(e.getCause().getCause()).isInstanceOf(IOException.class);
-			assertThat(e.getCause().getCause().getMessage()).isEqualTo("fail");
-		}
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> sync.synchronizeToLocalDirectory(mock(File.class)))
+				.withRootCauseInstanceOf(IOException.class)
+				.withMessageContaining("fail");
+
 		sync.synchronizeToLocalDirectory(mock(File.class));
 		assertThat(count.get()).isEqualTo(3);
 		sync.close();
@@ -261,6 +261,11 @@ public class AbstractRemoteFileSynchronizerTests {
 				return 0;
 			}
 
+			@Override
+			protected String protocol() {
+				return "file";
+			}
+
 			@Override
 			protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile,
 					File localDirectory, Session<String> session) {
@@ -269,13 +274,13 @@ public class AbstractRemoteFileSynchronizerTests {
 			}
 
 		};
-		sync.setFilter(new AcceptOnceFileListFilter<String>());
+		sync.setFilter(new AcceptOnceFileListFilter<>());
 		sync.setRemoteDirectory("foo");
 		sync.setBeanFactory(mock(BeanFactory.class));
 		return sync;
 	}
 
-	private class StringSessionFactory implements SessionFactory<String> {
+	private static class StringSessionFactory implements SessionFactory<String> {
 
 		@Override
 		public Session<String> getSession() {
@@ -284,7 +289,7 @@ public class AbstractRemoteFileSynchronizerTests {
 
 	}
 
-	private class StringSession implements Session<String> {
+	private static class StringSession implements Session<String> {
 
 		StringSession() {
 			super();
@@ -360,6 +365,11 @@ public class AbstractRemoteFileSynchronizerTests {
 			return null;
 		}
 
+		@Override
+		public String getHostPort() {
+			return null;
+		}
+
 	}
 
 }

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java

@@ -32,6 +32,7 @@ import org.springframework.integration.metadata.SimpleMetadataStore;
  * @author Mark Fisher
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 2.0
  */
 public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<FTPFile> {
@@ -62,4 +63,9 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<
 		return file.getTimestamp().getTimeInMillis();
 	}
 
+	@Override
+	protected String protocol() {
+		return "ftp";
+	}
+
 }

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java

@@ -38,13 +38,14 @@ import org.springframework.util.ObjectUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class FtpSession implements Session<FTPFile> {
 
-	private static final String SERVER_REPLIED_WITH = "'. Server replied with: ";
+	private static final Log LOGGER = LogFactory.getLog(FtpSession.class);
 
-	private final Log logger = LogFactory.getLog(this.getClass());
+	private static final String SERVER_REPLIED_WITH = "'. Server replied with: ";
 
 	private final FTPClient client;
 
@@ -86,7 +87,9 @@ public class FtpSession implements Session<FTPFile> {
 			throw new IOException("Failed to copy '" + path +
 					SERVER_REPLIED_WITH + this.client.getReplyString());
 		}
-		this.logger.info("File has been successfully transferred from: " + path);
+		if (LOGGER.isInfoEnabled()) {
+			LOGGER.info("File has been successfully transferred from: " + path);
+		}
 	}
 
 	@Override
@@ -109,8 +112,8 @@ public class FtpSession implements Session<FTPFile> {
 		}
 		if (this.client.completePendingCommand()) {
 			int replyCode = this.client.getReplyCode();
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug(this + " finalizeRaw - reply code: " + replyCode);
+			if (LOGGER.isDebugEnabled()) {
+				LOGGER.debug(this + " finalizeRaw - reply code: " + replyCode);
 			}
 			return FTPReply.isPositiveCompletion(replyCode);
 		}
@@ -126,8 +129,8 @@ public class FtpSession implements Session<FTPFile> {
 			throw new IOException("Failed to write to '" + path
 					+ SERVER_REPLIED_WITH + this.client.getReplyString());
 		}
-		if (this.logger.isInfoEnabled()) {
-			this.logger.info("File has been successfully transferred to: " + path);
+		if (LOGGER.isInfoEnabled()) {
+			LOGGER.info("File has been successfully transferred to: " + path);
 		}
 	}
 
@@ -140,8 +143,8 @@ public class FtpSession implements Session<FTPFile> {
 			throw new IOException("Failed to append to '" + path
 					+ SERVER_REPLIED_WITH + this.client.getReplyString());
 		}
-		if (this.logger.isInfoEnabled()) {
-			this.logger.info("File has been successfully appended to: " + path);
+		if (LOGGER.isInfoEnabled()) {
+			LOGGER.info("File has been successfully appended to: " + path);
 		}
 	}
 
@@ -150,17 +153,15 @@ public class FtpSession implements Session<FTPFile> {
 		try {
 			if (this.readingRaw.get()) {
 				if (!finalizeRaw()) {
-					if (this.logger.isWarnEnabled()) {
-						this.logger.warn("Finalize on readRaw() returned false for " + this);
+					if (LOGGER.isWarnEnabled()) {
+						LOGGER.warn("Finalize on readRaw() returned false for " + this);
 					}
 				}
 			}
 			this.client.disconnect();
 		}
 		catch (Exception e) {
-			if (this.logger.isWarnEnabled()) {
-				this.logger.warn("failed to disconnect FTPClient", e);
-			}
+			LOGGER.warn("failed to disconnect FTPClient", e);
 		}
 	}
 
@@ -183,8 +184,8 @@ public class FtpSession implements Session<FTPFile> {
 			throw new IOException("Failed to rename '" + pathFrom +
 					"' to " + pathTo + SERVER_REPLIED_WITH + this.client.getReplyString());
 		}
-		if (this.logger.isInfoEnabled()) {
-			this.logger.info("File has been successfully renamed from: " + pathFrom + " to " + pathTo);
+		if (LOGGER.isInfoEnabled()) {
+			LOGGER.info("File has been successfully renamed from: " + pathFrom + " to " + pathTo);
 		}
 	}
 
@@ -228,6 +229,10 @@ public class FtpSession implements Session<FTPFile> {
 		return this.client;
 	}
 
+	@Override
+	public String getHostPort() {
+		return this.client.getRemoteAddress().getHostName() + ':' + this.client.getRemotePort();
+	}
 
 	@Override
 	public boolean test() {

spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.2.xsd

@@ -177,6 +177,29 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="remote-file-metadata-store" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+											type="org.springframework.integration.metadata.MetadataStore" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								Reference to a MetadataStore for saving remote files information between
+								synchronization and polling.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="metadata-store-prefix" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Specify a prefix for metadata store to distinguish keys from another places
+								where the same shared store is used.
+								By default, the remote a component name is used.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attributeGroup ref="tempSuffixGroup" />
 				</xsd:extension>
 			</xsd:complexContent>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml

@@ -30,12 +30,16 @@
 				temporary-file-suffix=".foo"
 				max-fetch-size="42"
 				local-filter="acceptAllFilter"
-				remote-directory-expression="'foo/bar'">
+				remote-directory-expression="'foo/bar'"
+				remote-file-metadata-store="metadataStore"
+				metadata-store-prefix="testPrefix">
 			<int:poller fixed-rate="1000">
 				<int:transactional synchronization-factory="syncFactory"/>
 			</int:poller>
 	</int-ftp:inbound-channel-adapter>
 
+	<bean id="metadataStore" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
+
 	<bean id="dirScanner" class="org.springframework.integration.file.HeadDirectoryScanner">
 		<constructor-arg value="1" />
 	</bean>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java

@@ -48,6 +48,7 @@ import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
 import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
 import org.springframework.integration.ftp.session.FtpSession;
+import org.springframework.integration.metadata.MetadataStore;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
@@ -87,6 +88,9 @@ public class FtpInboundChannelAdapterParserTests {
 	@Autowired
 	private DirectoryScanner dirScanner;
 
+	@Autowired
+	private MetadataStore metadataStore;
+
 	@Test
 	public void testFtpInboundChannelAdapterComplete() throws Exception {
 		assertThat(TestUtils.getPropertyValue(ftpInbound, "autoStartup", Boolean.class)).isFalse();
@@ -109,6 +113,9 @@ public class FtpInboundChannelAdapterParserTests {
 		assertThat(TestUtils.getPropertyValue(fisync, "localFilenameGeneratorExpression")).isNotNull();
 		assertThat(TestUtils.getPropertyValue(fisync, "preserveTimestamp", Boolean.class)).isTrue();
 		assertThat(TestUtils.getPropertyValue(fisync, "temporaryFileSuffix", String.class)).isEqualTo(".foo");
+		assertThat(TestUtils.getPropertyValue(fisync, "remoteFileMetadataStore", MetadataStore.class))
+				.isSameAs(this.metadataStore);
+		assertThat(TestUtils.getPropertyValue(fisync, "metadataStorePrefix", String.class)).isEqualTo("testPrefix");
 		String remoteFileSeparator = (String) TestUtils.getPropertyValue(fisync, "remoteFileSeparator");
 		assertThat(remoteFileSeparator).isNotNull();
 		assertThat(remoteFileSeparator).isEqualTo("");
@@ -123,11 +130,11 @@ public class FtpInboundChannelAdapterParserTests {
 		assertThat(filtersIterator.next()).isInstanceOf(FtpPersistentAcceptOnceFileListFilter.class);
 
 		Object sessionFactory = TestUtils.getPropertyValue(fisync, "remoteFileTemplate.sessionFactory");
-		assertThat(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass())).isTrue();
+		assertThat(sessionFactory).isInstanceOf(DefaultFtpSessionFactory.class);
 		FileListFilter<?> acceptAllFilter = context.getBean("acceptAllFilter", FileListFilter.class);
 		assertThat(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class)
 				.contains(acceptAllFilter)).isTrue();
-		final AtomicReference<Method> genMethod = new AtomicReference<Method>();
+		final AtomicReference<Method> genMethod = new AtomicReference<>();
 		ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, method -> {
 			method.setAccessible(true);
 			genMethod.set(method);
@@ -137,10 +144,10 @@ public class FtpInboundChannelAdapterParserTests {
 	}
 
 	@Test
-	public void cachingSessionFactory() throws Exception {
+	public void cachingSessionFactory() {
 		Object sessionFactory = TestUtils.getPropertyValue(simpleAdapterWithCachedSessions,
 				"source.synchronizer.remoteFileTemplate.sessionFactory");
-		assertThat(sessionFactory.getClass()).isEqualTo(CachingSessionFactory.class);
+		assertThat(sessionFactory).isInstanceOf(CachingSessionFactory.class);
 		FtpInboundFileSynchronizer fisync =
 				TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.synchronizer",
 						FtpInboundFileSynchronizer.class);
@@ -161,7 +168,7 @@ public class FtpInboundChannelAdapterParserTests {
 	public static class TestSessionFactoryBean implements FactoryBean<DefaultFtpSessionFactory> {
 
 		@Override
-		public DefaultFtpSessionFactory getObject() throws Exception {
+		public DefaultFtpSessionFactory getObject() {
 			DefaultFtpSessionFactory factory = mock(DefaultFtpSessionFactory.class);
 			FtpSession session = mock(FtpSession.class);
 			when(factory.getSession()).thenReturn(session);

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java

@@ -98,6 +98,8 @@ public class FtpTests extends FtpTestSupport {
 		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
 		Message<?> message = out.receive(10_000);
 		assertThat(message).isNotNull();
+		assertThat(message.getHeaders())
+				.containsKeys(FileHeaders.REMOTE_HOST_PORT, FileHeaders.REMOTE_DIRECTORY, FileHeaders.REMOTE_FILE);
 		Object payload = message.getPayload();
 		assertThat(payload).isInstanceOf(File.class);
 		File file = (File) payload;
@@ -153,6 +155,7 @@ public class FtpTests extends FtpTestSupport {
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
 		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isIn(" ftpSource1.txt", "ftpSource2.txt");
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_HOST_PORT, String.class)).contains("localhost:");
 		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
 
 		message = out.receive(10_000);

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java

@@ -25,10 +25,12 @@ import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.OutputStream;
+import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
@@ -51,7 +53,9 @@ import org.springframework.integration.file.filters.RegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.session.AbstractFtpSessionFactory;
+import org.springframework.integration.metadata.MetadataStore;
 import org.springframework.integration.metadata.PropertiesPersistingMetadataStore;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 
@@ -60,6 +64,7 @@ import org.springframework.messaging.Message;
  * @author Gunnar Hillert
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class FtpInboundRemoteFileSystemSynchronizerTests {
@@ -76,7 +81,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 	public void testCopyFileToLocalDir() throws Exception {
 		File localDirectory = new File("test");
 		assertThat(localDirectory.exists()).isFalse();
-
+		MetadataStore remoteFileMetadataStore = new SimpleMetadataStore();
 		TestFtpSessionFactory ftpSessionFactory = new TestFtpSessionFactory();
 		ftpSessionFactory.setUsername("kermit");
 		ftpSessionFactory.setPassword("frog");
@@ -85,16 +90,18 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		synchronizer.setDeleteRemoteFiles(true);
 		synchronizer.setPreserveTimestamp(true);
 		synchronizer.setRemoteDirectory("remote-test-dir");
+		synchronizer.setRemoteFileMetadataStore(remoteFileMetadataStore);
+		synchronizer.setMetadataStorePrefix("ftpPollingTest:");
 		FtpRegexPatternFileListFilter patternFilter = new FtpRegexPatternFileListFilter(".*\\.test$");
 		PropertiesPersistingMetadataStore store = spy(new PropertiesPersistingMetadataStore());
 		store.setBaseDirectory("test");
 		store.afterPropertiesSet();
 		FtpPersistentAcceptOnceFileListFilter persistFilter =
 				new FtpPersistentAcceptOnceFileListFilter(store, "foo");
-		List<FileListFilter<FTPFile>> filters = new ArrayList<FileListFilter<FTPFile>>();
+		List<FileListFilter<FTPFile>> filters = new ArrayList<>();
 		filters.add(persistFilter);
 		filters.add(patternFilter);
-		CompositeFileListFilter<FTPFile> filter = new CompositeFileListFilter<FTPFile>(filters);
+		CompositeFileListFilter<FTPFile> filter = new CompositeFileListFilter<>(filters);
 		synchronizer.setFilter(filter);
 
 		ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
@@ -108,9 +115,9 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		ms.setAutoCreateLocalDirectory(true);
 		ms.setLocalDirectory(localDirectory);
 		ms.setBeanFactory(mock(BeanFactory.class));
-		CompositeFileListFilter<File> localFileListFilter = new CompositeFileListFilter<File>();
+		CompositeFileListFilter<File> localFileListFilter = new CompositeFileListFilter<>();
 		localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.TEST\\.a$"));
-		AcceptOnceFileListFilter<File> localAcceptOnceFilter = new AcceptOnceFileListFilter<File>();
+		AcceptOnceFileListFilter<File> localAcceptOnceFilter = new AcceptOnceFileListFilter<>();
 		localFileListFilter.addFilter(localAcceptOnceFilter);
 		RecursiveDirectoryScanner scanner = new RecursiveDirectoryScanner();
 		ms.setScanner(scanner);
@@ -143,8 +150,12 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 
 		TestUtils.getPropertyValue(localAcceptOnceFilter, "seenSet", Collection.class).clear();
 
-		new File("test/subdir/A.TEST.a").delete();
-		new File("test/subdir/B.TEST.a").delete();
+		File aFile = new File("test/subdir/A.TEST.a");
+		aFile.delete();
+		synchronizer.removeRemoteFileMetadata(aFile);
+		File bFile = new File("test/subdir/B.TEST.a");
+		bFile.delete();
+		synchronizer.removeRemoteFileMetadata(bFile);
 		// the remote filter should prevent a re-fetch
 		nothing = ms.receive();
 		assertThat(nothing).isNull();
@@ -152,11 +163,14 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		ms.stop();
 		verify(synchronizer).close();
 		verify(store).close();
+
+		Map<?, ?> metadata = TestUtils.getPropertyValue(remoteFileMetadataStore, "metadata", Map.class);
+		assertThat(metadata).isEmpty();
 	}
 
 
 	@Test
-	public void testSyncRemoteFileOnlyOnceByDefault() throws Exception {
+	public void testSyncRemoteFileOnlyOnceByDefault() {
 		File localDirectory = new File("test");
 		localDirectory.mkdir();
 
@@ -204,7 +218,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 
 	public static class TestFtpSessionFactory extends AbstractFtpSessionFactory<FTPClient> {
 
-		private final Collection<FTPFile> ftpFiles = new ArrayList<FTPFile>();
+		private final Collection<FTPFile> ftpFiles = new ArrayList<>();
 
 		private void init() {
 			String[] files = new File("remote-test-dir").list();
@@ -237,8 +251,10 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 							Mockito.any(OutputStream.class))).thenReturn(true);
 				}
 				when(ftpClient.listFiles("remote-test-dir"))
-						.thenReturn(ftpFiles.toArray(new FTPFile[ftpFiles.size()]));
+						.thenReturn(ftpFiles.toArray(new FTPFile[0]));
 				when(ftpClient.deleteFile(Mockito.anyString())).thenReturn(true);
+				when(ftpClient.getRemoteAddress()).thenReturn(InetAddress.getByName("localhost"));
+				when(ftpClient.getRemotePort()).thenReturn(-1);
 				return ftpClient;
 			}
 			catch (Exception e) {

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java

@@ -123,6 +123,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		received = (Message<byte[]>) this.data.receive(10000);
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO)).isInstanceOf(FtpFileInfo.class);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_HOST_PORT, String.class)).contains("localhost:");
 		assertThat(TestUtils.getPropertyValue(source, "toBeReceived", BlockingQueue.class)).hasSize(1);
 		assertThat(this.metadataMap).hasSize(1);
 		this.adapter.stop();

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java

@@ -27,6 +27,7 @@ import java.io.File;
 import java.io.FileOutputStream;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Calendar;
@@ -88,12 +89,12 @@ public class FtpOutboundTests {
 			file.delete();
 		}
 		assertThat(file.exists()).isFalse();
-		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(sessionFactory);
+		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<>(sessionFactory);
 		handler.setRemoteDirectoryExpression(new LiteralExpression("remote-target-dir"));
 		handler.setFileNameGenerator(message -> "handlerContent.test");
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
-		handler.handleMessage(new GenericMessage<String>("String data"));
+		handler.handleMessage(new GenericMessage<>("String data"));
 		assertThat(file.exists()).isTrue();
 		byte[] inFile = FileCopyUtils.copyToByteArray(file);
 		assertThat(new String(inFile)).isEqualTo("String data");
@@ -107,12 +108,12 @@ public class FtpOutboundTests {
 			file.delete();
 		}
 		assertThat(file.exists()).isFalse();
-		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(sessionFactory);
+		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<>(sessionFactory);
 		handler.setRemoteDirectoryExpression(new LiteralExpression("remote-target-dir"));
 		handler.setFileNameGenerator(message -> "handlerContent.test");
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
-		handler.handleMessage(new GenericMessage<byte[]>("byte[] data".getBytes()));
+		handler.handleMessage(new GenericMessage<>("byte[] data".getBytes()));
 		assertThat(file.exists()).isTrue();
 		byte[] inFile = FileCopyUtils.copyToByteArray(file);
 		assertThat(new String(inFile)).isEqualTo("byte[] data");
@@ -124,7 +125,7 @@ public class FtpOutboundTests {
 		File targetDir = new File("remote-target-dir");
 		assertThat(targetDir.exists()).as("target directory does not exist: " + targetDir.getName()).isTrue();
 
-		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(sessionFactory);
+		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<>(sessionFactory);
 		handler.setRemoteDirectoryExpression(new LiteralExpression(targetDir.getName()));
 		handler.setFileNameGenerator(message -> ((File) message.getPayload()).getName() + ".test");
 		handler.setBeanFactory(mock(BeanFactory.class));
@@ -141,7 +142,7 @@ public class FtpOutboundTests {
 	}
 
 	@Test
-	public void testHandleMissingFileMessage() throws Exception {
+	public void testHandleMissingFileMessage() {
 		File targetDir = new File("remote-target-dir");
 		assertThat(targetDir.exists()).as("target directory does not exist: " + targetDir.getName()).isTrue();
 
@@ -191,7 +192,7 @@ public class FtpOutboundTests {
 	}
 
 	@Test //INT-2275
-	public void testFtpOutboundGatewayInsideChain() throws Exception {
+	public void testFtpOutboundGatewayInsideChain() {
 		ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
 				"FtpOutboundInsideChainTests-context.xml", getClass());
 
@@ -248,7 +249,9 @@ public class FtpOutboundTests {
 							any(OutputStream.class))).thenReturn(true);
 				}
 				when(ftpClient.listFiles("remote-test-dir/"))
-						.thenReturn(ftpFiles.toArray(new FTPFile[ftpFiles.size()]));
+						.thenReturn(ftpFiles.toArray(new FTPFile[0]));
+				when(ftpClient.getRemoteAddress())
+						.thenReturn(InetAddress.getByName("127.0.0.1"));
 				return ftpClient;
 			}
 			catch (Exception e) {

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java

@@ -59,4 +59,9 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer
 		return (long) file.getAttrs().getMTime() * 1000;
 	}
 
+	@Override
+	protected String protocol() {
+		return "sftp";
+	}
+
 }

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java

@@ -46,16 +46,18 @@ import com.jcraft.jsch.SftpException;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class SftpSession implements Session<LsEntry> {
 
+	private static final Log LOGGER = LogFactory.getLog(SftpSession.class);
+
 	private static final String SESSION_IS_NOT_CONNECTED = "session is not connected";
 
 	private static final Duration DEFAULT_CHANNEL_CONNECT_TIMEOUT = Duration.ofSeconds(5);
 
-	private final Log logger = LogFactory.getLog(this.getClass());
-
 	private final com.jcraft.jsch.Session jschSession;
 
 	private final JSchSessionWrapper wrapper;
@@ -214,14 +216,14 @@ public class SftpSession implements Session<LsEntry> {
 			this.channel.rename(pathFrom, pathTo);
 		}
 		catch (SftpException sftpex) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: "
-						+ pathTo + " and execute rename again.");
+			if (LOGGER.isDebugEnabled()) {
+				LOGGER.debug("Initial File rename failed, possibly because file already exists. " +
+						"Will attempt to delete file: " + pathTo + " and execute rename again.");
 			}
 			try {
-				this.remove(pathTo);
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug("Delete file: " + pathTo + " succeeded. Will attempt rename again");
+				remove(pathTo);
+				if (LOGGER.isDebugEnabled()) {
+					LOGGER.debug("Delete file: " + pathTo + " succeeded. Will attempt rename again");
 				}
 			}
 			catch (IOException ioex) {
@@ -240,8 +242,8 @@ public class SftpSession implements Session<LsEntry> {
 				throw exception; // NOSONAR - added to suppressed exceptions
 			}
 		}
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug("File: " + pathFrom + " was successfully renamed to " + pathTo);
+		if (LOGGER.isDebugEnabled()) {
+			LOGGER.debug("File: " + pathFrom + " was successfully renamed to " + pathTo);
 		}
 	}
 
@@ -300,6 +302,11 @@ public class SftpSession implements Session<LsEntry> {
 		return this.channel;
 	}
 
+	@Override
+	public String getHostPort() {
+		return this.jschSession.getHost() + ':' + this.jschSession.getPort();
+	}
+
 	@Override
 	public boolean test() {
 		return isOpen() && doTest();

spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.2.xsd

@@ -180,6 +180,29 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="remote-file-metadata-store" type="xsd:string">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+											type="org.springframework.integration.metadata.MetadataStore" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								Reference to a MetadataStore for saving remote files information between
+								synchronization and polling.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="metadata-store-prefix" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Specify a prefix for metadata store to distinguish keys from another places
+								where the same shared store is used.
+								By default, the remote a component name is used.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attributeGroup ref="tempSuffixGroup" />
 				</xsd:extension>
 			</xsd:complexContent>

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml

@@ -45,12 +45,16 @@
 								  max-fetch-size="42"
 								  delete-remote-files="${delete.remote.files}"
 								  auto-startup="false"
-								  preserve-timestamp="true">
+								  preserve-timestamp="true"
+								  remote-file-metadata-store="metadataStore"
+								  metadata-store-prefix="testPrefix">
 		<poller fixed-rate="1000">
 			<transactional synchronization-factory="syncFactory"/>
 		</poller>
 	</sftp:inbound-channel-adapter>
 
+	<beans:bean id="metadataStore" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
+
 	<beans:bean id="acceptAllFilter" class="org.springframework.integration.file.filters.AcceptAllFileListFilter"/>
 
 	<transaction-synchronization-factory id="syncFactory">

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.sftp.config;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 import java.io.File;
 import java.util.Collection;
@@ -34,6 +35,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.metadata.MetadataStore;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource;
 import org.springframework.integration.test.util.TestUtils;
@@ -44,6 +46,7 @@ import org.springframework.messaging.PollableChannel;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 public class InboundChannelAdapterParserTests {
 
@@ -53,9 +56,9 @@ public class InboundChannelAdapterParserTests {
 	}
 
 	@Test
-	public void testAutoStartup() throws Exception {
+	public void testAutoStartup() {
 		ConfigurableApplicationContext context =
-			new ClassPathXmlApplicationContext("SftpInboundAutostartup-context.xml", this.getClass());
+				new ClassPathXmlApplicationContext("SftpInboundAutostartup-context.xml", this.getClass());
 
 		SourcePollingChannelAdapter adapter = context.getBean("sftpAutoStartup", SourcePollingChannelAdapter.class);
 		assertThat(adapter.isRunning()).isFalse();
@@ -63,15 +66,15 @@ public class InboundChannelAdapterParserTests {
 	}
 
 	@Test
-	public void testWithLocalFiles() throws Exception {
+	public void testWithLocalFiles() {
 		ConfigurableApplicationContext context =
-			new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
+				new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
 		assertThat(new File("src/main/resources").exists()).isTrue();
 
 		Object adapter = context.getBean("sftpAdapterAutoCreate");
 		assertThat(adapter instanceof SourcePollingChannelAdapter).isTrue();
 		SftpInboundFileSynchronizingMessageSource source =
-			(SftpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(adapter, "source");
+				(SftpInboundFileSynchronizingMessageSource) TestUtils.getPropertyValue(adapter, "source");
 		assertThat(source).isNotNull();
 
 		PriorityBlockingQueue<?> blockingQueue =
@@ -87,6 +90,10 @@ public class InboundChannelAdapterParserTests {
 		assertThat(TestUtils.getPropertyValue(synchronizer, "preserveTimestamp", Boolean.class)).isTrue();
 		String remoteFileSeparator = (String) TestUtils.getPropertyValue(synchronizer, "remoteFileSeparator");
 		assertThat(TestUtils.getPropertyValue(synchronizer, "temporaryFileSuffix", String.class)).isEqualTo(".bar");
+		assertThat(TestUtils.getPropertyValue(synchronizer, "remoteFileMetadataStore"))
+				.isSameAs(context.getBean(MetadataStore.class));
+		assertThat(TestUtils.getPropertyValue(synchronizer, "metadataStorePrefix", String.class))
+				.isEqualTo("testPrefix");
 		assertThat(remoteFileSeparator).isNotNull();
 		assertThat(remoteFileSeparator).isEqualTo(".");
 		PollableChannel requestChannel = context.getBean("requestChannel", PollableChannel.class);
@@ -103,7 +110,7 @@ public class InboundChannelAdapterParserTests {
 	@Test
 	public void testAutoChannel() {
 		ConfigurableApplicationContext context =
-			new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
+				new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context.xml", this.getClass());
 		// Auto-created channel
 		MessageChannel autoChannel = context.getBean("autoChannel", MessageChannel.class);
 		SourcePollingChannelAdapter autoChannelAdapter = context.getBean("autoChannel.adapter",
@@ -117,30 +124,36 @@ public class InboundChannelAdapterParserTests {
 		context.close();
 	}
 
-	@Test(expected = BeanDefinitionStoreException.class)
+	@Test
 	//exactly one of 'filename-pattern' or 'filter' is allowed on SFTP inbound adapter
-	public void testFailWithFilePatternAndFilter() throws Exception {
+	public void testFailWithFilePatternAndFilter() {
 		assertThat(!new File("target/bar").exists()).isTrue();
-		new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context-fail.xml", this.getClass()).close();
+		assertThatExceptionOfType(BeanDefinitionStoreException.class)
+				.isThrownBy(() ->
+						new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context-fail.xml",
+								getClass()));
 	}
 
 	@Test
-	public void testLocalDirAutoCreated() throws Exception {
+	public void testLocalDirAutoCreated() {
 		assertThat(new File("foo").exists()).isFalse();
 		ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
-				"InboundChannelAdapterParserTests-context.xml", this.getClass());
+				"InboundChannelAdapterParserTests-context.xml", getClass());
 		assertThat(new File("foo").exists()).isTrue();
 		context.close();
 	}
 
-	@Test(expected = BeanCreationException.class)
-	public void testLocalDirAutoCreateFailed() throws Exception {
-		new ClassPathXmlApplicationContext("InboundChannelAdapterParserTests-context-fail-autocreate.xml",
-				this.getClass()).close();
+	@Test
+	public void testLocalDirAutoCreateFailed() {
+		assertThatExceptionOfType(BeanCreationException.class)
+				.isThrownBy(() ->
+						new ClassPathXmlApplicationContext(
+								"InboundChannelAdapterParserTests-context-fail-autocreate.xml",
+								getClass()));
 	}
 
 	@After
-	public void cleanUp() throws Exception {
+	public void cleanUp() {
 		new File("foo").delete();
 	}
 

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java

@@ -118,6 +118,7 @@ public class SftpTests extends SftpTestSupport {
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload()).isInstanceOf(InputStream.class);
 		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE)).isIn(" sftpSource1.txt", "sftpSource2.txt");
+		assertThat(message.getHeaders().get(FileHeaders.REMOTE_HOST_PORT, String.class)).contains("localhost:");
 		((InputStream) message.getPayload()).close();
 		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
 

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

@@ -36,6 +36,7 @@ import org.junit.Before;
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
@@ -58,6 +59,7 @@ import com.jcraft.jsch.SftpATTRS;
  * @author Gunnar Hillert
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class SftpInboundRemoteFileSystemSynchronizerTests {
@@ -97,7 +99,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 		store.afterPropertiesSet();
 		SftpPersistentAcceptOnceFileListFilter persistFilter =
 				new SftpPersistentAcceptOnceFileListFilter(store, "foo");
-		List<FileListFilter<LsEntry>> filters = new ArrayList<FileListFilter<LsEntry>>();
+		List<FileListFilter<LsEntry>> filters = new ArrayList<>();
 		filters.add(persistFilter);
 		filters.add(patternFilter);
 		CompositeFileListFilter<LsEntry> filter = new CompositeFileListFilter<LsEntry>(filters);
@@ -109,27 +111,30 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 		ms.setAutoCreateLocalDirectory(true);
 		ms.setLocalDirectory(localDirectory);
 		ms.setBeanFactory(mock(BeanFactory.class));
-		CompositeFileListFilter<File> localFileListFilter = new CompositeFileListFilter<File>();
+		CompositeFileListFilter<File> localFileListFilter = new CompositeFileListFilter<>();
 		localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.test$"));
-		AcceptOnceFileListFilter<File> localAcceptOnceFilter = new AcceptOnceFileListFilter<File>();
+		AcceptOnceFileListFilter<File> localAcceptOnceFilter = new AcceptOnceFileListFilter<>();
 		localFileListFilter.addFilter(localAcceptOnceFilter);
 		ms.setLocalFilter(localFileListFilter);
 		ms.afterPropertiesSet();
 		ms.start();
 
-		Message<File> atestFile =  ms.receive();
+		Message<File> atestFile = ms.receive();
 		assertThat(atestFile).isNotNull();
 		assertThat(atestFile.getPayload().getName()).isEqualTo("a.test");
 		// The test remote files are created with the current timestamp + 1 day.
 		assertThat(atestFile.getPayload().lastModified()).isGreaterThan(System.currentTimeMillis());
 
-		Message<File> btestFile =  ms.receive();
+		assertThat(atestFile.getHeaders())
+				.containsKeys(FileHeaders.REMOTE_HOST_PORT, FileHeaders.REMOTE_DIRECTORY, FileHeaders.REMOTE_FILE);
+
+		Message<File> btestFile = ms.receive();
 		assertThat(btestFile).isNotNull();
 		assertThat(btestFile.getPayload().getName()).isEqualTo("b.test");
 		// The test remote files are created with the current timestamp + 1 day.
 		assertThat(atestFile.getPayload().lastModified()).isGreaterThan(System.currentTimeMillis());
 
-		Message<File> nothing =  ms.receive();
+		Message<File> nothing = ms.receive();
 		assertThat(nothing).isNull();
 
 		// two times because on the third receive (above) the internal queue will be empty, so it will attempt
@@ -143,7 +148,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 		new File("test/a.test").delete();
 		new File("test/b.test").delete();
 		// the remote filter should prevent a re-fetch
-		nothing =  ms.receive();
+		nothing = ms.receive();
 		assertThat(nothing).isNull();
 
 		ms.stop();
@@ -153,7 +158,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 
 	public static class TestSftpSessionFactory extends DefaultSftpSessionFactory {
 
-		private final Vector<LsEntry> sftpEntries = new Vector<LsEntry>();
+		private final Vector<LsEntry> sftpEntries = new Vector<>();
 
 		private void init() {
 			String[] files = new File("remote-test-dir").list();

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java

@@ -119,6 +119,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		received = (Message<byte[]>) this.data.receive(10000);
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO)).isInstanceOf(SftpFileInfo.class);
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_HOST_PORT, String.class)).contains("localhost:");
 		this.adapter.stop();
 	}
 

src/reference/asciidoc/ftp.adoc

@@ -444,7 +444,7 @@ the removal of the failed file from the filter:
 <int-ftp:inbound-channel-adapter id="ftpAdapter"
         session-factory="ftpSessionFactory"
         channel="requestChannel"
-        remote-directory-expression="'/sftpSource'"
+        remote-directory-expression="'/ftpSource'"
         local-directory="file:myLocalDir"
         auto-create-local-directory="true"
         filename-pattern="*.txt">
@@ -636,6 +636,7 @@ Starting with version 5.0, the `FileHeaders.REMOTE_FILE_INFO` header provides ad
 If you set the `fileInfoJson` property on the `FtpStreamingMessageSource` to `false`, the header contains an `FtpFileInfo` object.
 The `FTPFile` object provided by the underlying Apache Net library can be accessed by using the `FtpFileInfo.getFileInfo()` method.
 The `fileInfoJson` property is not available when you use XML configuration, but you can set it by injecting the `FtpStreamingMessageSource` into one of your configuration classes.
+See also <<ftp-remote-file-info>>.
 
 Starting with version 5.1, the generic type of the `comparator` is `FTPFile`.
 Previously, it was `AbstractFileInfo<FTPFile>`.
@@ -1575,3 +1576,19 @@ public ApplicationEventListeningMessageProducer eventsAdapter() {
 }
 ----
 ====
+
+[[ftp-remote-file-info]]
+=== Remote File Information
+
+Starting with version 5.2, the `FtpStreamingMessageSource` (<<ftp-streaming>>), `FtpInboundFileSynchronizingMessageSource` (<<ftp-inbound>>) and "read"-commands of the `FtpOutboundGateway` (<<ftp-outbound-gateway>>) provide additional headers in the message to produce with an information about the remote file:
+
+* `FileHeaders.REMOTE_HOST_PORT` - the host:port pair the remote session has been connected to during file transfer operation;
+* `FileHeaders.REMOTE_DIRECTORY` - the remote directory the operation has been performed;
+* `FileHeaders.REMOTE_FILE` - the remote file name; applicable only for single file operations.
+
+Since the `FtpInboundFileSynchronizingMessageSource` doesn't produce messages against remote files, but using a local copy, the `AbstractInboundFileSynchronizer` stores an information about remote file in the `MetadataStore` (which can be configured externally) in the URI style (`protocol://host:port/remoteDirectory#remoteFileName`) during synchronization operation.
+This metadata is retrieved by the `FtpInboundFileSynchronizingMessageSource` when local file is polled.
+When local file is deleted, it is recommended to remove its metadata entry.
+The `AbstractInboundFileSynchronizer` provides a `removeRemoteFileMetadata()` callback for this purpose.
+In addition there is a `setMetadataStorePrefix()` to be used in the metadata keys.
+It is recommended to have this prefix be different from the one used in the `MetadataStore`-based `FileListFilter` implementations, when the same `MetadataStore` instance is shared between these components, to avoid entry overriding because both filter and `AbstractInboundFileSynchronizer` use the same local file name for the metadata entry key.

src/reference/asciidoc/sftp.adoc

@@ -200,7 +200,7 @@ injection, as the following example shows:
 
 Version 4.2 introduced the `DelegatingSessionFactory`, which allows the selection of the actual session factory at
 runtime.
-Prior to invoking the FTP endpoint, you can call `setThreadKey()` on the factory to associate a key with the current thread.
+Prior to invoking the SFTP endpoint, you can call `setThreadKey()` on the factory to associate a key with the current thread.
 That key is then used to look up the actual session factory to be used.
 You can clear the key by calling `clearThreadKey()` after use.
 
@@ -334,7 +334,7 @@ Unlike outbound gateways and adapters, where the root object of the SpEL evaluat
 Consequently, the root object of the SpEL evaluation context is the original name of the remote file (a `String`).
 
 The inbound channel adapter first retrieves the file to a local directory and then emits each file according to the poller configuration.
-Starting with version 5.0, you can limit the number of files fetched from the FTP server when new file retrievals are needed.
+Starting with version 5.0, you can limit the number of files fetched from the SFTP server when new file retrievals are needed.
 This can be beneficial when the target files are large or when running in a clustered system with a persistent file list filter, discussed later in this section.
 Use `max-fetch-size` for this purpose.
 A negative value (the default) means no limit and all matching files are retrieved.
@@ -363,7 +363,7 @@ Since version 4.0, this filter requires a `ConcurrentMetadataStore`.
 When used with a shared data store (such as `Redis` with the `RedisMetadataStore`), this lets filter keys be shared across multiple application or server instances.
 
 Starting with version 5.0, the `SftpPersistentAcceptOnceFileListFilter` with an in-memory `SimpleMetadataStore` is applied by default for the `SftpInboundFileSynchronizer`.
-This filter is also applied, together with the `regex` or `pattern` option in the XML configuration, as well as through `FtpInboundChannelAdapterSpec` in Java DSL.
+This filter is also applied, together with the `regex` or `pattern` option in the XML configuration, as well as through `SftpInboundChannelAdapterSpec` in Java DSL.
 You can handle any other use-cases by using `CompositeFileListFilter` (or `ChainFileListFilter`).
 
 The above discussion refers to filtering the files before retrieving them.
@@ -635,6 +635,7 @@ Starting with version 5.0, the `FileHeaders.REMOTE_FILE_INFO` header provides ad
 If you set the `fileInfoJson` property on the `SftpStreamingMessageSource` to `false`, the header contains an `SftpFileInfo` object.
 You can access the `LsEntry` object provided by the underlying Jsch library by using the `SftpFileInfo.getFileInfo()` method.
 The `fileInfoJson` property is not available when you use XML configuration, but you can set it by injecting the `SftpStreamingMessageSource` into one of your configuration classes.
+See also <<sftp-remote-file-info>>.
 
 Starting with version 5.1, the generic type of the `comparator` is `LsEntry`.
 Previously, it was `AbstractFileInfo<LsEntry>`.
@@ -771,8 +772,8 @@ This allows files retrieved from different directories to be downloaded to simil
 ----
 @Bean
 public IntegrationFlow flow() {
-    return IntegrationFlows.from(Ftp.inboundAdapter(sf())
-                    .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
+    return IntegrationFlows.from(Sftp.inboundAdapter(sf())
+                    .filter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
                     .localDirectory(new File(tmpDir))
                     .localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root")
                     .remoteDirectory("."),
@@ -1283,7 +1284,7 @@ public class SftpJavaApplication {
 
     @Bean
     public SessionFactory<LsEntry> sftpSessionFactory() {
-        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
+        DefaultSftpSessionFactory sf = new DefaultSftpSessionFactory();
         sf.setHost("localhost");
         sf.setPort(port);
         sf.setUsername("foo");
@@ -1319,8 +1320,7 @@ public class SftpJavaApplication {
 
 When performing operations on multiple files (by using `mget` and `mput`) an exception can occur some time after one or more files have been transferred.
 In this case (starting with version 4.2), a `PartialSuccessException` is thrown.
-As well as the usual `MessagingException` properties (`failedMessage` and `cause`), this exception has two additional
-properties:
+As well as the usual `MessagingException` properties (`failedMessage` and `cause`), this exception has two additional properties:
 
 * `partialResults`: The successful transfer results.
 * `derivedInput`: The list of files generated from the request message (such as local files to transfer for an `mput`).
@@ -1366,7 +1366,7 @@ log4j.category.com.jcraft.jsch=DEBUG
 === MessageSessionCallback
 
 Starting with Spring Integration version 4.2, you can use a `MessageSessionCallback<F, T>` implementation with the `<int-sftp:outbound-gateway/>` (`SftpOutboundGateway`) to perform any operation on the `Session<LsEntry>` with the `requestMessage` context.
-You can use it for any non-standard or low-level FTP operation (or several), such as allowing access from an integration flow definition, or functional interface (lambda) implementation injection.
+You can use it for any non-standard or low-level SFTP operation (or several), such as allowing access from an integration flow definition, or functional interface (lambda) implementation injection.
 The following example uses a lambda:
 
 ====
@@ -1433,3 +1433,19 @@ public ApplicationEventListeningMessageProducer eventsAdapter() {
 }
 ----
 ====
+
+[[sftp-remote-file-info]]
+=== Remote File Information
+
+Starting with version 5.2, the `SftpStreamingMessageSource` (<<sftp-streaming>>), `SftpInboundFileSynchronizingMessageSource` (<<sftp-inbound>>) and "read"-commands of the `SftpOutboundGateway` (<<sftp-outbound-gateway>>) provide additional headers in the message to produce with an information about the remote file:
+
+* `FileHeaders.REMOTE_HOST_PORT` - the host:port pair the remote session has been connected to during file transfer operation;
+* `FileHeaders.REMOTE_DIRECTORY` - the remote directory the operation has been performed;
+* `FileHeaders.REMOTE_FILE` - the remote file name; applicable only for single file operations.
+
+Since the `SftpInboundFileSynchronizingMessageSource` doesn't produce messages against remote files, but using a local copy, the `AbstractInboundFileSynchronizer` stores an information about remote file in the `MetadataStore` (which can be configured externally) in the URI style (`protocol://host:port/remoteDirectory#remoteFileName`) during synchronization operation.
+This metadata is retrieved by the `SftpInboundFileSynchronizingMessageSource` when local file is polled.
+When local file is deleted, it is recommended to remove its metadata entry.
+The `AbstractInboundFileSynchronizer` provides a `removeRemoteFileMetadata()` callback for this purpose.
+In addition there is a `setMetadataStorePrefix()` to be used in the metadata keys.
+It is recommended to have this prefix be different from the one used in the `MetadataStore`-based `FileListFilter` implementations, when the same `MetadataStore` instance is shared between these components, to avoid entry overriding because both filter and `AbstractInboundFileSynchronizer` use the same local file name for the metadata entry key.

src/reference/asciidoc/whats-new.adoc

@@ -157,4 +157,8 @@ See <<./router.adoc#dynamic-routers, Dynamic Routers>> for more information.
 ==== FTP/SFTP Changes
 
 The `RotatingServerAdvice` is decoupled now from the `RotationPolicy` and its `StandardRotationPolicy`.
-See <<./ftp.adoc#ftp-rotating-server-advice, Polling Multiple Servers and Directories>> for more information.
+
+The remote file information, including host/port and directory are included now into message headers in the `AbstractInboundFileSynchronizingMessageSource` and `AbstractRemoteFileStreamingMessageSource` implementations.
+Also this information is included into headers in the read operations results of the `AbstractRemoteFileOutboundGateway` implementations.
+
+See <<./ftp.adoc#ftp, FTP(S) Support>> and <<./sftp.adoc#sftp, SFTP Support>> for more information.