From a756e6334dcb442c90f6c1bdbd754c20293ef55b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 28 Aug 2019 08:58:14 -0400 Subject: [PATCH] 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 --- .../integration/file/FileHeaders.java | 5 + ...RemoteFileInboundChannelAdapterParser.java | 15 ++- .../RemoteFileInboundChannelAdapterSpec.java | 27 +++- ...actPersistentAcceptOnceFileListFilter.java | 36 +++--- ...tractRemoteFileStreamingMessageSource.java | 1 + .../AbstractRemoteFileOutboundGateway.java | 104 ++++++++------- .../remote/session/CachingSessionFactory.java | 36 +++--- .../file/remote/session/Session.java | 7 ++ .../AbstractInboundFileSynchronizer.java | 96 +++++++++++++- ...InboundFileSynchronizingMessageSource.java | 16 ++- .../RemoteFileOutboundGatewayTests.java | 118 +++++++----------- .../session/CachingSessionFactoryTests.java | 65 +++++----- .../AbstractRemoteFileSynchronizerTests.java | 40 +++--- .../inbound/FtpInboundFileSynchronizer.java | 6 + .../integration/ftp/session/FtpSession.java | 37 +++--- .../ftp/config/spring-integration-ftp-5.2.xsd | 23 ++++ ...boundChannelAdapterParserTests-context.xml | 6 +- .../FtpInboundChannelAdapterParserTests.java | 17 ++- .../integration/ftp/dsl/FtpTests.java | 3 + ...oundRemoteFileSystemSynchronizerTests.java | 36 ++++-- .../FtpStreamingMessageSourceTests.java | 1 + .../ftp/outbound/FtpOutboundTests.java | 19 +-- .../inbound/SftpInboundFileSynchronizer.java | 5 + .../integration/sftp/session/SftpSession.java | 27 ++-- .../config/spring-integration-sftp-5.2.xsd | 23 ++++ ...boundChannelAdapterParserTests-context.xml | 6 +- .../InboundChannelAdapterParserTests.java | 45 ++++--- .../integration/sftp/dsl/SftpTests.java | 1 + ...oundRemoteFileSystemSynchronizerTests.java | 21 ++-- .../SftpStreamingMessageSourceTests.java | 1 + src/reference/asciidoc/ftp.adoc | 19 ++- src/reference/asciidoc/sftp.adoc | 34 +++-- src/reference/asciidoc/whats-new.adoc | 6 +- 33 files changed, 605 insertions(+), 297 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java index 3486fb2e5c..cb924aef61 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileHeaders.java +++ b/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"; + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java index 013fe54540..6f488dab51 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java +++ b/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> getRegexPatternFileListFilterClass(); - protected abstract Class> getPersistentAcceptOnceFileListFilterClass(); + protected abstract Class> + getPersistentAcceptOnceFileListFilterClass(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java index 69e6d22426..9c7e4d3701 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java +++ b/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 getComponentsToRegister() { Map componentsToRegister = new LinkedHashMap<>(); componentsToRegister.put(this.synchronizer, null); - if (this.expressionFileListFilter != null) { componentsToRegister.put(this.expressionFileListFilter, null); } - return componentsToRegister; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java index 0df7b370f4..e159ee6d2f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java +++ b/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 extends AbstractFileListFilter - implements ReversibleFileListFilter, ResettableFileListFilter, Closeable { + implements ReversibleFileListFilter, ResettableFileListFilter, 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 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 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); } /** diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index 9ed74e25cc..4ba84d7b8e 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -205,6 +205,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource .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); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 3822eb6692..b055211e5d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/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 extends AbstractReplyProducingMessageHandler { - private final RemoteFileTemplate remoteFileTemplate; private final Command command; @@ -118,7 +117,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply public AbstractRemoteFileOutboundGateway(SessionFactory sessionFactory, MessageSessionCallback messageSessionCallback) { - this(new RemoteFileTemplate(sessionFactory), messageSessionCallback); + this(new RemoteFileTemplate<>(sessionFactory), messageSessionCallback); } /** @@ -161,7 +160,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply public AbstractRemoteFileOutboundGateway(SessionFactory sessionFactory, Command command, @Nullable String expression) { - this(new RemoteFileTemplate(sessionFactory), command, expression); + this(new RemoteFileTemplate<>(sessionFactory), command, expression); } /** @@ -317,7 +316,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply * @since 4.3 */ public void setRenameExpression(Expression renameExpression) { - this.renameProcessor = new ExpressionEvaluatingMessageProcessor(renameExpression); + this.renameProcessor = new ExpressionEvaluatingMessageProcessor<>(renameExpression); } /** @@ -490,10 +489,14 @@ public abstract class AbstractRemoteFileOutboundGateway 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 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 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 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 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 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 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 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) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index db37ae1aeb..e5c8cee7f7 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/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 implements SessionFactory, DisposableBean { - private static final Log logger = LogFactory.getLog(CachingSessionFactory.class); + private static final Log LOGGER = LogFactory.getLog(CachingSessionFactory.class); private final SessionFactory sessionFactory; @@ -55,7 +57,6 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe /** * Create a CachingSessionFactory with an unlimited number of sessions. - * * @param sessionFactory the underlying session factory. */ public CachingSessionFactory(SessionFactory sessionFactory) { @@ -66,11 +67,9 @@ public class CachingSessionFactory implements SessionFactory, 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. - *

- * Do not cache a {@link DelegatingSessionFactory}, cache each delegate therein instead. + *

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 implements SessionFactory, DisposableBe Assert.isTrue(!(sessionFactory instanceof DelegatingSessionFactory), "'sessionFactory' cannot be a 'DelegatingSessionFactory'; cache each delegate instead"); this.sessionFactory = sessionFactory; - this.pool = new SimplePool>(sessionCacheSize, new SimplePool.PoolItemCallback>() { + this.pool = new SimplePool<>(sessionCacheSize, new SimplePool.PoolItemCallback>() { @Override public Session createForPool() { return CachingSessionFactory.this.sessionFactory.getSession(); @@ -100,7 +99,6 @@ public class CachingSessionFactory implements SessionFactory, 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 implements SessionFactory, 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 implements SessionFactory, 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 implements SessionFactory, DisposableBe this.pool.removeAllIdleItems(); } - public class CachedSession implements Session { //NOSONAR (final) + public class CachedSession implements Session { //NOSONAR must be final, but can't for mocking in tests private final Session targetSession; @@ -190,17 +185,17 @@ public class CachingSessionFactory implements SessionFactory, 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 implements SessionFactory, DisposableBe return this.targetSession.getClientInstance(); } + @Override + public String getHostPort() { + return this.targetSession.getHostPort(); + } + } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java index a91397db1b..2bc1d5fce5 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java @@ -108,6 +108,13 @@ public interface Session 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()}. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index f10a9c4399..b2a39c22ae 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/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 - 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 /** * The current evaluation of the expression. */ - private volatile String evaluatedRemoteDirectory; + private String evaluatedRemoteDirectory; /** * An {@link FileListFilter} that runs against the remote file system view. @@ -124,9 +130,14 @@ public abstract class AbstractInboundFileSynchronizer @Nullable private Comparator 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 sessionFactory) { @@ -250,11 +261,37 @@ public abstract class AbstractInboundFileSynchronizer 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 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 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 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 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 } } + /** + * 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(); + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index a975284ae6..dae464a101 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/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 private volatile boolean running; - public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer synchronizer) { this(synchronizer, null); } @@ -192,7 +193,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource 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 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 Arrays.asList(this.localFileListFilter, new RegexPatternFileListFilter(completePattern))); } - /** * The {@link FileReadingMessageSource} extension to increase visibility * for the {@link FileReadingMessageSource#doReceive()} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java index bb835999f6..2a3f1e6362 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java +++ b/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 out; - try { - out = (MessageBuilder) 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) 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) gw.handleRequestMessage(new GenericMessage<>("f1")); @@ -652,7 +637,8 @@ public class RemoteFileOutboundGatewayTests { @Test public void testGetTempFileDelete() { - SessionFactory sessionFactory = mock(SessionFactory.class); + @SuppressWarnings("unchecked") + SessionFactory 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 sessionFactory = mock(SessionFactory.class); - @SuppressWarnings("unchecked") Session session = mock(Session.class); RemoteFileTemplate template = new RemoteFileTemplate<>(sessionFactory); template.setRemoteDirectoryExpression(new LiteralExpression("foo/")); @@ -897,7 +875,6 @@ public class RemoteFileOutboundGatewayTests { tempFolder.newFile("qux.txt"); Message requestMessage = MessageBuilder.withPayload(tempFolder.getRoot()) .build(); - @SuppressWarnings("unchecked") List out = (List) 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 sessionFactory = mock(SessionFactory.class); - @SuppressWarnings("unchecked") Session session = mock(Session.class); RemoteFileTemplate template = new RemoteFileTemplate<>(sessionFactory); template.setRemoteDirectoryExpression(new LiteralExpression("foo/")); @@ -931,7 +907,6 @@ public class RemoteFileOutboundGatewayTests { Message requestMessage = MessageBuilder.withPayload(tempFolder.getRoot()) .build(); - @SuppressWarnings("unchecked") List out = (List) 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 sessionFactory = mock(SessionFactory.class); - @SuppressWarnings("unchecked") Session session = mock(Session.class); RemoteFileTemplate template = new RemoteFileTemplate<>(sessionFactory); template.setRemoteDirectoryExpression(new LiteralExpression("foo/")); @@ -963,7 +937,6 @@ public class RemoteFileOutboundGatewayTests { files.add(tempFolder.newFile("buz.txt")); Message> requestMessage = MessageBuilder.withPayload(files) .build(); - @SuppressWarnings("unchecked") List out = (List) 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; + } + } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java index f388b0070b..eb9e148dfc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java +++ b/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 cache = new CachingSessionFactory(factory); + CachingSessionFactory cache = new CachingSessionFactory<>(factory); cache.setTestSession(true); Session 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 ccf = new CachingSessionFactory(factory); - RemoteFileTemplate template = new RemoteFileTemplate(ccf); + CachingSessionFactory ccf = new CachingSessionFactory<>(factory); + RemoteFileTemplate template = new RemoteFileTemplate<>(ccf); template.setFileNameExpression(new LiteralExpression("foo")); template.setBeanFactory(mock(BeanFactory.class)); template.afterPropertiesSet(); - try { - template.get(new GenericMessage("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 { + private static class TestSessionFactory implements SessionFactory { private int n; @@ -112,7 +112,7 @@ public class CachingSessionFactoryTests { } - private class TestSession implements Session { + private static class TestSession implements Session { @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; diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java index aae3ac77df..36eea32837 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java +++ b/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 session) throws IOException { @@ -83,19 +88,14 @@ public class AbstractRemoteFileSynchronizerTests { } }; - sync.setFilter(new AcceptOnceFileListFilter()); + 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 session) { @@ -269,13 +274,13 @@ public class AbstractRemoteFileSynchronizerTests { } }; - sync.setFilter(new AcceptOnceFileListFilter()); + sync.setFilter(new AcceptOnceFileListFilter<>()); sync.setRemoteDirectory("foo"); sync.setBeanFactory(mock(BeanFactory.class)); return sync; } - private class StringSessionFactory implements SessionFactory { + private static class StringSessionFactory implements SessionFactory { @Override public Session getSession() { @@ -284,7 +289,7 @@ public class AbstractRemoteFileSynchronizerTests { } - private class StringSession implements Session { + private static class StringSession implements Session { StringSession() { super(); @@ -360,6 +365,11 @@ public class AbstractRemoteFileSynchronizerTests { return null; } + @Override + public String getHostPort() { + return null; + } + } } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java index 23c427e83c..8ff99dc9c1 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java +++ b/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 { @@ -62,4 +63,9 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< return file.getTimestamp().getTimeInMillis(); } + @Override + protected String protocol() { + return "ftp"; + } + } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index 7cd8f8e932..a72baaf00a 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/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 { - 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 { 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 { } 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 { 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 { 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 { 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 { 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 { return this.client; } + @Override + public String getHostPort() { + return this.client.getRemoteAddress().getHostName() + ':' + this.client.getRemotePort(); + } @Override public boolean test() { diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.2.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.2.xsd index 2620fab5c7..ac7f47d01f 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.2.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.2.xsd @@ -177,6 +177,29 @@ + + + + + + + + + Reference to a MetadataStore for saving remote files information between + synchronization and polling. + + + + + + + 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. + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml index 8c1b8e20f6..1c14275be0 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml +++ b/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"> + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index a29e99c16d..93e8b793da 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/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 genMethod = new AtomicReference(); + final AtomicReference 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 { @Override - public DefaultFtpSessionFactory getObject() throws Exception { + public DefaultFtpSessionFactory getObject() { DefaultFtpSessionFactory factory = mock(DefaultFtpSessionFactory.class); FtpSession session = mock(FtpSession.class); when(factory.getSession()).thenReturn(session); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java index 59f054c2c3..02bdc48c0a 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java +++ b/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); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java index 38117006d1..3ae08eb510 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java +++ b/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> filters = new ArrayList>(); + List> filters = new ArrayList<>(); filters.add(persistFilter); filters.add(patternFilter); - CompositeFileListFilter filter = new CompositeFileListFilter(filters); + CompositeFileListFilter 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 localFileListFilter = new CompositeFileListFilter(); + CompositeFileListFilter localFileListFilter = new CompositeFileListFilter<>(); localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.TEST\\.a$")); - AcceptOnceFileListFilter localAcceptOnceFilter = new AcceptOnceFileListFilter(); + AcceptOnceFileListFilter 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 { - private final Collection ftpFiles = new ArrayList(); + private final Collection 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) { diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java index 5b8accbcc0..75d6b72efd 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -123,6 +123,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { received = (Message) 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(); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java index 2247a28e5e..b64cac5f14 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java +++ b/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 handler = new FileTransferringMessageHandler(sessionFactory); + FileTransferringMessageHandler 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 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 handler = new FileTransferringMessageHandler(sessionFactory); + FileTransferringMessageHandler 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[] 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 handler = new FileTransferringMessageHandler(sessionFactory); + FileTransferringMessageHandler 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) { diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java index 37ca321e36..245ee20ffb 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java +++ b/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"; + } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index aee02da36b..8b8509d707 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/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 { + 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 { 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 { 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 { return this.channel; } + @Override + public String getHostPort() { + return this.jschSession.getHost() + ':' + this.jschSession.getPort(); + } + @Override public boolean test() { return isOpen() && doTest(); diff --git a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.2.xsd b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.2.xsd index 41be25e27e..7982ade06d 100644 --- a/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.2.xsd +++ b/spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.2.xsd @@ -180,6 +180,29 @@ + + + + + + + + + Reference to a MetadataStore for saving remote files information between + synchronization and polling. + + + + + + + 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. + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml index c82c6013ab..5ac5b65261 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml +++ b/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"> + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java index 0319e4e827..41495f6c26 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java +++ b/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(); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java index 0fc965560f..e4a9e011d7 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java +++ b/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(); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java index 055539db7c..82f2845cd8 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/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> filters = new ArrayList>(); + List> filters = new ArrayList<>(); filters.add(persistFilter); filters.add(patternFilter); CompositeFileListFilter filter = new CompositeFileListFilter(filters); @@ -109,27 +111,30 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { ms.setAutoCreateLocalDirectory(true); ms.setLocalDirectory(localDirectory); ms.setBeanFactory(mock(BeanFactory.class)); - CompositeFileListFilter localFileListFilter = new CompositeFileListFilter(); + CompositeFileListFilter localFileListFilter = new CompositeFileListFilter<>(); localFileListFilter.addFilter(new RegexPatternFileListFilter(".*\\.test$")); - AcceptOnceFileListFilter localAcceptOnceFilter = new AcceptOnceFileListFilter(); + AcceptOnceFileListFilter localAcceptOnceFilter = new AcceptOnceFileListFilter<>(); localFileListFilter.addFilter(localAcceptOnceFilter); ms.setLocalFilter(localFileListFilter); ms.afterPropertiesSet(); ms.start(); - Message atestFile = ms.receive(); + Message 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 btestFile = ms.receive(); + assertThat(atestFile.getHeaders()) + .containsKeys(FileHeaders.REMOTE_HOST_PORT, FileHeaders.REMOTE_DIRECTORY, FileHeaders.REMOTE_FILE); + + Message 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 nothing = ms.receive(); + Message 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 sftpEntries = new Vector(); + private final Vector sftpEntries = new Vector<>(); private void init() { String[] files = new File("remote-test-dir").list(); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index 5de32e0b7d..9a47bd9e95 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -119,6 +119,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { received = (Message) 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(); } diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 4f78fac03e..1355add3b3 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -444,7 +444,7 @@ the removal of the failed file from the filter: @@ -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 <>. Starting with version 5.1, the generic type of the `comparator` is `FTPFile`. Previously, it was `AbstractFileInfo`. @@ -1575,3 +1576,19 @@ public ApplicationEventListeningMessageProducer eventsAdapter() { } ---- ==== + +[[ftp-remote-file-info]] +=== Remote File Information + +Starting with version 5.2, the `FtpStreamingMessageSource` (<>), `FtpInboundFileSynchronizingMessageSource` (<>) and "read"-commands of the `FtpOutboundGateway` (<>) 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. diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 385615047c..a0c21b66e5 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/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 <>. Starting with version 5.1, the generic type of the `comparator` is `LsEntry`. Previously, it was `AbstractFileInfo`. @@ -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 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` implementation with the `` (`SftpOutboundGateway`) to perform any operation on the `Session` 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` (<>), `SftpInboundFileSynchronizingMessageSource` (<>) and "read"-commands of the `SftpOutboundGateway` (<>) 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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 0bd024e27b..1f4497b326 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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.