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 a13b7a3625..4d83bed689 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -102,11 +102,6 @@ public abstract class AbstractInboundFileSynchronizer */ private Expression remoteDirectoryExpression; - /** - * The current evaluation of the expression. - */ - private String evaluatedRemoteDirectory; - /** * An {@link FileListFilter} that runs against the remote file system view. */ @@ -203,7 +198,6 @@ public abstract class AbstractInboundFileSynchronizer */ public void setRemoteDirectory(String remoteDirectory) { this.remoteDirectoryExpression = new LiteralExpression(remoteDirectory); - evaluateRemoteDirectory(); } /** @@ -229,7 +223,6 @@ public abstract class AbstractInboundFileSynchronizer protected final void doSetRemoteDirectoryExpression(Expression expression) { Assert.notNull(expression, "'remoteDirectoryExpression' must not be null"); this.remoteDirectoryExpression = expression; - evaluateRemoteDirectory(); } /** @@ -298,7 +291,6 @@ public abstract class AbstractInboundFileSynchronizer if (this.evaluationContext == null) { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory); } - evaluateRemoteDirectory(); if (!StringUtils.hasText(this.metadataStorePrefix)) { this.metadataStorePrefix = this.name; } @@ -341,26 +333,27 @@ public abstract class AbstractInboundFileSynchronizer } return; } + String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class); if (this.logger.isTraceEnabled()) { - this.logger.trace("Synchronizing " + this.evaluatedRemoteDirectory + " to " + localDirectory); + this.logger.trace("Synchronizing " + remoteDirectory + " to " + localDirectory); } try { int transferred = this.remoteFileTemplate.execute(session -> - transferFilesFromRemoteToLocal(localDirectory, maxFetchSize, session)); + transferFilesFromRemoteToLocal(remoteDirectory, localDirectory, maxFetchSize, session)); if (this.logger.isDebugEnabled()) { - this.logger.debug(transferred + " files transferred from '" + this.evaluatedRemoteDirectory + "'"); + this.logger.debug(transferred + " files transferred from '" + remoteDirectory + "'"); } } catch (Exception e) { throw new MessagingException("Problem occurred while synchronizing '" - + this.evaluatedRemoteDirectory + "' to local directory", e); + + remoteDirectory + "' to local directory", e); } } - private Integer transferFilesFromRemoteToLocal(File localDirectory, int maxFetchSize, Session session) - throws IOException { + private Integer transferFilesFromRemoteToLocal(String remoteDirectory, File localDirectory, + int maxFetchSize, Session session) throws IOException { - F[] files = session.list(this.evaluatedRemoteDirectory); + F[] files = session.list(remoteDirectory); if (!ObjectUtils.isEmpty(files)) { files = FileUtils.purgeUnwantedElements(files, e -> !isFile(e), this.comparator); } @@ -372,6 +365,12 @@ public abstract class AbstractInboundFileSynchronizer int copied = filteredFiles.size(); int accepted = 0; + EvaluationContext localFileEvaluationContext = null; + if (this.localFilenameGeneratorExpression != null) { + localFileEvaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory); + localFileEvaluationContext.setVariable("remoteDirectory", remoteDirectory); + } + for (F file : filteredFiles) { if (filteringOneByOne) { if ((maxFetchSize < 0 || accepted < maxFetchSize) && this.filter @@ -383,7 +382,9 @@ public abstract class AbstractInboundFileSynchronizer copied--; } } - copied = copyIfNotNull(localDirectory, session, filteringOneByOne, filteredFiles, copied, file); + copied = + copyIfNotNull(remoteDirectory, localDirectory, localFileEvaluationContext, session, + filteringOneByOne, filteredFiles, copied, file); } return copied; } @@ -392,13 +393,16 @@ public abstract class AbstractInboundFileSynchronizer } } - private int copyIfNotNull(File localDirectory, Session session, boolean filteringOneByOne, List filteredFiles, - int copied, @Nullable F file) throws IOException { + private int copyIfNotNull(String remoteDirectory, File localDirectory, + @Nullable EvaluationContext localFileEvaluationContext, Session session, boolean filteringOneByOne, + List filteredFiles, int copied, @Nullable F file) throws IOException { boolean renamedFailed = false; try { - if (file != null && !copyFileToLocalDirectory(this.evaluatedRemoteDirectory, file, - localDirectory, session)) { + if (file != null && + !copyFileToLocalDirectory(remoteDirectory, localFileEvaluationContext, file, localDirectory, + session)) { + renamedFailed = true; } } @@ -440,11 +444,12 @@ public abstract class AbstractInboundFileSynchronizer } } - protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, // NOSONAR - File localDirectory, Session session) throws IOException { + protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, // NOSONAR + @Nullable EvaluationContext localFileEvaluationContext, F remoteFile, File localDirectory, + Session session) throws IOException { String remoteFileName = getFilename(remoteFile); - String localFileName = generateLocalFileName(remoteFileName); + String localFileName = generateLocalFileName(remoteFileName, localFileEvaluationContext); String remoteFilePath = remoteDirectoryPath != null ? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName) : remoteFileName; @@ -567,22 +572,16 @@ public abstract class AbstractInboundFileSynchronizer return renamed; } - private String generateLocalFileName(String remoteFileName) { + private String generateLocalFileName(String remoteFileName, + @Nullable EvaluationContext localFileEvaluationContext) { + if (this.localFilenameGeneratorExpression != null) { - return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName, + return this.localFilenameGeneratorExpression.getValue(localFileEvaluationContext, remoteFileName, String.class); } return remoteFileName; } - protected void evaluateRemoteDirectory() { - if (this.evaluationContext != null) { - this.evaluatedRemoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, - String.class); - this.evaluationContext.setVariable("remoteDirectory", this.evaluatedRemoteDirectory); - } - } - /** * Obtain a metadata for remote file associated with the provided local file. * @param localFile the local file to retrieve metadata for. 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 5c73e12878..ddc1e293c9 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,21 +18,27 @@ package org.springframework.integration.file.remote.synchronizer; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import static org.mockito.Mockito.mock; import java.io.File; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.LinkedList; +import java.util.Queue; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; -import org.junit.Test; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.springframework.beans.factory.BeanFactory; +import org.springframework.expression.EvaluationContext; +import org.springframework.integration.expression.SupplierExpression; import org.springframework.integration.file.HeadDirectoryScanner; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.ChainFileListFilter; @@ -78,8 +84,10 @@ public class AbstractRemoteFileSynchronizerTests { } @Override - protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, + protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, + EvaluationContext localFileEvaluationContext, String remoteFile, File localDirectory, Session session) throws IOException { + if ("bar".equals(remoteFile) && failWhenCopyingBar.getAndSet(false)) { throw new IOException("fail"); } @@ -209,13 +217,60 @@ public class AbstractRemoteFileSynchronizerTests { assertThat(count.get()).isEqualTo(1); } - @Test(expected = IllegalStateException.class) + @Test public void testScannerAndWatchServiceConflict() { final AtomicInteger count = new AtomicInteger(); AbstractInboundFileSynchronizingMessageSource source = createSource(count); source.setUseWatchService(true); source.setScanner(new HeadDirectoryScanner(1)); - source.afterPropertiesSet(); + assertThatIllegalStateException() + .isThrownBy(source::afterPropertiesSet); + } + + @Test + public void testRemoteDirectoryRefreshedOnEachSynchronization(@TempDir File localDir) { + AbstractInboundFileSynchronizer sync = + new AbstractInboundFileSynchronizer(new StringSessionFactory()) { + + @Override + protected boolean isFile(String file) { + return true; + } + + @Override + protected String getFilename(String file) { + return file; + } + + @Override + protected long getModified(String file) { + return 0; + } + + @Override + protected String protocol() { + return "mock"; + } + + }; + + Queue remoteDirs = new LinkedList<>(); + remoteDirs.add("dir1"); + remoteDirs.add("dir2"); + sync.setRemoteDirectoryExpression(new SupplierExpression<>(remoteDirs::poll)); + sync.setLocalFilenameGeneratorExpressionString("#remoteDirectory+'/'+#root"); + sync.setBeanFactory(mock(BeanFactory.class)); + sync.afterPropertiesSet(); + + sync.synchronizeToLocalDirectory(localDir); + sync.synchronizeToLocalDirectory(localDir); + + /*Files.find(localDir.toPath(), + Integer.MAX_VALUE, + (filePath, fileAttr) -> fileAttr.isRegularFile()) + .forEach(System.out::println);*/ + + assertThat(localDir.list()).contains("dir1", "dir2"); } private AbstractInboundFileSynchronizingMessageSource createSource(AtomicInteger count) { @@ -267,8 +322,10 @@ public class AbstractRemoteFileSynchronizerTests { } @Override - protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, + protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, + EvaluationContext localFileEvaluationContext, String remoteFile, File localDirectory, Session session) { + count.incrementAndGet(); return true; } @@ -301,7 +358,7 @@ public class AbstractRemoteFileSynchronizerTests { @Override public String[] list(String path) { - return new String[] { "foo", "bar", "baz" }; + return new String[]{ "foo", "bar", "baz" }; } @Override @@ -366,7 +423,7 @@ public class AbstractRemoteFileSynchronizerTests { @Override public String getHostPort() { - return null; + return "mock:6666"; } } 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 66b23009f9..68c0e42515 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -36,6 +36,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.expression.Expression; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.file.DirectoryScanner; import org.springframework.integration.file.filters.CompositeFileListFilter; import org.springframework.integration.file.filters.FileListFilter; @@ -136,7 +137,8 @@ public class FtpInboundChannelAdapterParserTests { method.setAccessible(true); genMethod.set(method); }, method -> "generateLocalFileName".equals(method.getName())); - assertThat(genMethod.get().invoke(fisync, "foo")).isEqualTo("FOO.afoo"); + assertThat(genMethod.get().invoke(fisync, "foo", ExpressionUtils.createStandardEvaluationContext(this.context))) + .isEqualTo("FOO.afoo"); assertThat(inbound.getMaxFetchSize()).isEqualTo(42); }