diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java index b3eb49ea71..baa7dba2d8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2018 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. @@ -27,9 +27,10 @@ import org.springframework.messaging.Message; * should be ignored and/or take action after the receive. * * @author Gary Russell + * * @since 4.2 */ -public abstract class AbstractMessageSourceAdvice implements MethodInterceptor { +public abstract class AbstractMessageSourceAdvice implements MethodInterceptor, MessageSourceMutator { @Override public final Object invoke(MethodInvocation invocation) throws Throwable { @@ -45,20 +46,4 @@ public abstract class AbstractMessageSourceAdvice implements MethodInterceptor { return afterReceive(result, (MessageSource) target); } - /** - * Subclasses can decide whether to proceed with this poll. - * @param source the message source. - * @return true to proceed. - */ - public abstract boolean beforeReceive(MessageSource source); - - /** - * Subclasses can take actions based on the result of the poll; e.g. - * adjust the {@code trigger}. The message can also be replaced with a new one. - * @param result the received message. - * @param source the message source. - * @return a message to continue to process the result, null to discard whatever the poll returned. - */ - public abstract Message afterReceive(Message result, MessageSource source); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java index f23288708a..9bd3891a0e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -47,11 +47,6 @@ public class CompoundTriggerAdvice extends AbstractMessageSourceAdvice { this.override = overrideTrigger; } - @Override - public boolean beforeReceive(MessageSource source) { - return true; - } - @Override public Message afterReceive(Message result, MessageSource source) { if (result == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessageSourceMutator.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessageSourceMutator.java new file mode 100644 index 0000000000..1936943b58 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessageSourceMutator.java @@ -0,0 +1,52 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aop; + +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.Message; + +/** + * An object that can mutate a {@link MessageSource} before and/or after + * {@link MessageSource#receive()} is called. + * + * @author Gary Russell + * + * @since 5.0.7. + * + */ +@FunctionalInterface +public interface MessageSourceMutator { + + /** + * Subclasses can decide whether to proceed with this poll. + * @param source the message source. + * @return true to proceed (default). + */ + default boolean beforeReceive(MessageSource source) { + return true; + } + + /** + * Subclasses can take actions based on the result of the poll; e.g. + * adjust the {@code trigger}. The message can also be replaced with a new one. + * @param result the received message. + * @param source the message source. + * @return a message to continue to process the result, null to discard whatever the poll returned. + */ + Message afterReceive(Message result, MessageSource source); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java index 0696d0db7f..b5ec20846a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2018 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. @@ -61,11 +61,6 @@ public class SimpleActiveIdleMessageSourceAdvice extends AbstractMessageSourceAd this.activePollPeriod = activePollPeriod; } - @Override - public boolean beforeReceive(MessageSource source) { - return true; - } - @Override public Message afterReceive(Message result, MessageSource source) { if (result == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java index 70a04d31a7..5678da6373 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java @@ -75,7 +75,7 @@ public final class PollerSpec extends IntegrationComponentSpec adviceChain; + private boolean errorHandlerIsDefault; - private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); + private Trigger trigger = new PeriodicTrigger(10); + + private List adviceChain; + + private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); + + private long maxMessagesPerPoll = -1; + + private TransactionSynchronizationFactory transactionSynchronizationFactory; private volatile ScheduledFuture runningTask; @@ -78,18 +86,23 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private volatile boolean initialized; - private volatile long maxMessagesPerPoll = -1; - - private final Object initializationMonitor = new Object(); - - private volatile TransactionSynchronizationFactory transactionSynchronizationFactory; - public AbstractPollingEndpoint() { this.setPhase(Integer.MAX_VALUE / 2); } public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = (taskExecutor != null ? taskExecutor : new SyncTaskExecutor()); + this.syncExecutor = this.taskExecutor instanceof SyncTaskExecutor + || (this.taskExecutor instanceof ErrorHandlingTaskExecutor + && ((ErrorHandlingTaskExecutor) this.taskExecutor).isSyncExecutor()); + } + + protected Executor getTaskExecutor() { + return this.taskExecutor; + } + + protected boolean isSyncExecutor() { + return this.syncExecutor; } public void setTrigger(Trigger trigger) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 01e9f2eea9..ccbc15ac32 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -27,7 +27,7 @@ import org.springframework.aop.support.AopUtils; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.Lifecycle; -import org.springframework.integration.aop.AbstractMessageSourceAdvice; +import org.springframework.integration.aop.MessageSourceMutator; import org.springframework.integration.context.ExpressionCapable; import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.MessagingTemplate; @@ -137,7 +137,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint @Override protected boolean isReceiveOnlyAdvice(Advice advice) { - return advice instanceof AbstractMessageSourceAdvice; + return advice instanceof MessageSourceMutator; } @Override @@ -159,6 +159,13 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint } this.appliedAdvices.clear(); this.appliedAdvices.addAll(chain); + if (!(isSyncExecutor()) && logger.isWarnEnabled()) { + logger.warn(getComponentName() + ": A task executor is supplied and " + chain.size() + + "MessageSourceMutator(s) is/are provided. If an advice mutates the source, such " + + "mutations are not thread safe and could cause unexpected results, especially with " + + "high frequency pollers. Consider using a downstream ExecutorChannel instead of " + + "adding an executor to the poller"); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java index 8f1f5b80f5..ecec5296b5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,6 +18,7 @@ package org.springframework.integration.util; import java.util.concurrent.Executor; +import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; @@ -45,6 +46,9 @@ public class ErrorHandlingTaskExecutor implements TaskExecutor { this.errorHandler = errorHandler; } + public boolean isSyncExecutor() { + return this.executor instanceof SyncTaskExecutor; + } @Override public void execute(final Runnable task) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/RotatingServerAdvice.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/RotatingServerAdvice.java new file mode 100644 index 0000000000..f0905b10cd --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/RotatingServerAdvice.java @@ -0,0 +1,247 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.remote.session; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.aop.AbstractMessageSourceAdvice; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource; +import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * A smart poller advice that rotates across multiple remote servers/directories. + * + * @author Gary Russell + * + * @since 5.0.7 + * + */ +public class RotatingServerAdvice extends AbstractMessageSourceAdvice { + + private final RotationPolicy rotationPolicy; + + /** + * Create an instance that rotates to the next server/directory if no message is + * received. + * @param factory the {@link DelegatingSessionFactory}. + * @param keyDirectories a list of {@link KeyDirectory}. + */ + public RotatingServerAdvice(DelegatingSessionFactory factory, List keyDirectories) { + this(factory, keyDirectories, false); + } + + /** + * Create an instance that rotates to the next server/directory depending on the fair + * argument. + * @param factory the {@link DelegatingSessionFactory}. + * @param keyDirectories a list of {@link KeyDirectory}. + * @param fair true to rotate on every poll, false to rotate when no message is received. + */ + public RotatingServerAdvice(DelegatingSessionFactory factory, List keyDirectories, boolean fair) { + this(new StandardRotationPolicy(factory, keyDirectories, fair)); + } + + /** + * Construct an instance that rotates according to the supplied + * {@link RotationPolicy}. + * @param rotationPolicy the policy. + */ + public RotatingServerAdvice(RotationPolicy rotationPolicy) { + Assert.notNull(rotationPolicy, "'rotationPolicy' cannot be null"); + this.rotationPolicy = rotationPolicy; + } + + @Override + public boolean beforeReceive(MessageSource source) { + this.rotationPolicy.beforeReceive(source); + return true; + } + + @Override + public Message afterReceive(Message result, MessageSource source) { + this.rotationPolicy.afterReceive(result != null, source); + return result; + } + + /** + * Implementations can reconfigure the message source before and/or after + * a poll. + */ + public interface RotationPolicy { + + /** + * Invoked before the message source receive() method. + * @param source the message source. + */ + void beforeReceive(MessageSource source); + + /** + * Invoked after the message source receive() method. + * @param messageReceived true if a message was received. + * @param source the message source. + */ + void afterReceive(boolean messageReceived, MessageSource source); + + } + + /** + * Standard rotation policy; iterates over key/directory pairs; when the end + * is reached, starts again at the beginning. If the fair option is true + * the rotation occurs on every poll, regardless of result. Otherwise rotation + * occurs when the current pair returns no message. + */ + public static class StandardRotationPolicy implements RotationPolicy { + + protected final Log logger = LogFactory.getLog(getClass()); + + private final DelegatingSessionFactory factory; + + private final List keyDirectories = new ArrayList<>(); + + private final boolean fair; + + private volatile Iterator iterator; + + private volatile KeyDirectory current; + + private volatile boolean initialized; + + public StandardRotationPolicy(DelegatingSessionFactory factory, List keyDirectories, + boolean fair) { + + Assert.notNull(factory, "factory cannot be null"); + Assert.notNull(keyDirectories, "keyDirectories cannot be null"); + Assert.isTrue(keyDirectories.size() > 0, "At least one KeyDirectory is required"); + this.factory = factory; + this.keyDirectories.addAll(keyDirectories); + this.fair = fair; + this.iterator = this.keyDirectories.iterator(); + } + + protected Iterator getIterator() { + return this.iterator; + } + + protected void setIterator(Iterator iterator) { + this.iterator = iterator; + } + + protected boolean isInitialized() { + return this.initialized; + } + + protected void setInitialized(boolean initialized) { + this.initialized = initialized; + } + + protected DelegatingSessionFactory getFactory() { + return this.factory; + } + + protected List getKeyDirectories() { + return this.keyDirectories; + } + + protected boolean isFair() { + return this.fair; + } + + @Override + public void beforeReceive(MessageSource source) { + if (this.fair || !this.initialized) { + configureSource(source); + this.initialized = true; + } + if (this.logger.isTraceEnabled()) { + this.logger.trace("Next poll is for " + this.current); + } + this.factory.setThreadKey(this.current.getKey()); + } + + @Override + public void afterReceive(boolean messageReceived, MessageSource source) { + if (this.logger.isTraceEnabled()) { + this.logger.trace("Poll produced " + + (messageReceived ? "a" : "no") + + " message"); + } + this.factory.clearThreadKey(); + if (!this.fair && !messageReceived) { + configureSource(source); + } + } + + protected void configureSource(MessageSource source) { + Assert.isTrue(source instanceof AbstractInboundFileSynchronizingMessageSource + || source instanceof AbstractRemoteFileStreamingMessageSource, + "source must be an AbstractInboundFileSynchronizingMessageSource or a " + + "AbstractRemoteFileStreamingMessageSource"); + if (!this.iterator.hasNext()) { + this.iterator = this.keyDirectories.iterator(); + } + this.current = this.iterator.next(); + if (source instanceof AbstractRemoteFileStreamingMessageSource) { + ((AbstractRemoteFileStreamingMessageSource) source).setRemoteDirectory(this.current.getDirectory()); + } + else { + ((AbstractInboundFileSynchronizingMessageSource) source).getSynchronizer() + .setRemoteDirectory(this.current.getDirectory()); + } + } + + } + + /** + * A {@link DelegatingSessionFactory} key/directory pair. + */ + public static class KeyDirectory { + + private final String key; + + private final String directory; + + public KeyDirectory(String key, String directory) { + Assert.notNull(key, "key cannot be null"); + Assert.notNull(directory, "directory cannot be null"); + this.key = key; + this.directory = directory; + } + + public String getKey() { + return this.key; + } + + public String getDirectory() { + return this.directory; + } + + @Override + public String toString() { + return "KeyDirectory [key=" + this.key + ", directory=" + this.directory + "]"; + } + + } + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SessionFactoryMapBuilder.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SessionFactoryMapBuilder.java new file mode 100644 index 0000000000..5654c060fc --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SessionFactoryMapBuilder.java @@ -0,0 +1,30 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.file.remote.session; + +import org.springframework.integration.support.MapBuilder; + +/** + * A {@link MapBuilder} to producer a map that maps objects to {@link SessionFactory}s. + * + * @author Gary Russell + * @since 5.0.7 + * + */ +public class SessionFactoryMapBuilder extends MapBuilder, Object, SessionFactory> { + +} 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 5e87690360..364674af7c 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-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -75,38 +75,43 @@ public abstract class AbstractInboundFileSynchronizer private final RemoteFileTemplate remoteFileTemplate; - private volatile EvaluationContext evaluationContext; + private EvaluationContext evaluationContext; - private volatile String remoteFileSeparator = "/"; + private String remoteFileSeparator = "/"; /** * Extension used when downloading files. We change it right after we know it's downloaded. */ - private volatile String temporaryFileSuffix = ".writing"; + private String temporaryFileSuffix = ".writing"; - private volatile Expression localFilenameGeneratorExpression; + private Expression localFilenameGeneratorExpression; /** * the path on the remote mount as a String. */ private volatile Expression remoteDirectoryExpression; + /** + * The current evaluation of the expression. + */ + private volatile String evaluatedRemoteDirectory; + /** * An {@link FileListFilter} that runs against the remote file system view. */ - private volatile FileListFilter filter; + private FileListFilter filter; /** * Should we delete the remote source files * after copying to the local directory? By default this is false. */ - private volatile boolean deleteRemoteFiles; + private boolean deleteRemoteFiles; /** * Should we transfer the remote file timestamp * to the local file? By default this is false. */ - private volatile boolean preserveTimestamp; + private boolean preserveTimestamp; private BeanFactory beanFactory; @@ -164,6 +169,7 @@ public abstract class AbstractInboundFileSynchronizer */ public void setRemoteDirectory(String remoteDirectory) { this.remoteDirectoryExpression = new LiteralExpression(remoteDirectory); + evaluateRemoteDirectory(); } /** @@ -182,13 +188,14 @@ public abstract class AbstractInboundFileSynchronizer * @see #setRemoteDirectoryExpression(Expression) */ public void setRemoteDirectoryExpressionString(String remoteDirectoryExpression) { - setRemoteDirectoryExpression(EXPRESSION_PARSER.parseExpression(remoteDirectoryExpression)); + doSetRemoteDirectoryExpression(EXPRESSION_PARSER.parseExpression(remoteDirectoryExpression)); } protected final void doSetRemoteDirectoryExpression(Expression remoteDirectoryExpression) { Assert.notNull(remoteDirectoryExpression, "'remoteDirectoryExpression' must not be null"); this.remoteDirectoryExpression = remoteDirectoryExpression; + evaluateRemoteDirectory(); } /** @@ -231,9 +238,11 @@ public abstract class AbstractInboundFileSynchronizer if (this.evaluationContext == null) { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory); } + evaluateRemoteDirectory(); doInit(); } + /** * Subclasses can override to perform initialization - called from * {@link InitializingBean#afterPropertiesSet()}. @@ -269,10 +278,12 @@ public abstract class AbstractInboundFileSynchronizer } return; } - final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class); + if (this.logger.isTraceEnabled()) { + this.logger.trace("Synchronizing " + this.evaluatedRemoteDirectory + " to " + localDirectory); + } try { int transferred = this.remoteFileTemplate.execute(session -> { - F[] files = session.list(remoteDirectory); + F[] files = session.list(this.evaluatedRemoteDirectory); if (!ObjectUtils.isEmpty(files)) { List filteredFiles = filterFiles(files); if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { @@ -288,10 +299,9 @@ public abstract class AbstractInboundFileSynchronizer for (F file : filteredFiles) { try { - if (file != null) { - if (!copyFileToLocalDirectory(remoteDirectory, file, localDirectory, session)) { - copied--; - } + if (file != null && !copyFileToLocalDirectory(this.evaluatedRemoteDirectory, file, + localDirectory, session)) { + copied--; } } catch (RuntimeException e1) { @@ -328,8 +338,8 @@ public abstract class AbstractInboundFileSynchronizer protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory, Session session) throws IOException { - String remoteFileName = this.getFilename(remoteFile); - String localFileName = this.generateLocalFileName(remoteFileName); + String remoteFileName = getFilename(remoteFile); + String localFileName = generateLocalFileName(remoteFileName); String remoteFilePath = remoteDirectoryPath != null ? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName) : remoteFileName; @@ -438,11 +448,20 @@ public abstract class AbstractInboundFileSynchronizer private String generateLocalFileName(String remoteFileName) { if (this.localFilenameGeneratorExpression != null) { - return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName, String.class); + return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, 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); + } + } + protected abstract boolean isFile(F file); protected abstract String getFilename(F file); 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 ed9121fb94..d73e658eeb 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 @@ -164,6 +164,15 @@ public abstract class AbstractInboundFileSynchronizingMessageSource this.scannerExplicitlySet = true; } + /** + * Return the underlying synchronizer. + * @return the synchronizer. + * @since 5.0.7 + */ + public AbstractInboundFileSynchronizer getSynchronizer() { + return this.synchronizer; + } + @Override public void afterPropertiesSet() throws Exception { super.afterPropertiesSet(); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java index 5030e70eea..5c58da79ff 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpTestSupport.java @@ -81,14 +81,17 @@ public class FtpTestSupport extends RemoteFileTestSupport { } public static SessionFactory sessionFactory() { + return new CachingSessionFactory(rawSessionFactory()); + } + + protected static DefaultFtpSessionFactory rawSessionFactory() { DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory(); sf.setHost("localhost"); sf.setPort(port); sf.setUsername("foo"); sf.setPassword("foo"); sf.setClientMode(FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE); - - return new CachingSessionFactory(sf); + return sf; } private static class TestUserManager implements UserManager { 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 8b2498c60d..f6f288d674 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-2017 the original author or authors. + * Copyright 2002-2018 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. diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java new file mode 100644 index 0000000000..c2da6d2d10 --- /dev/null +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/RotatingServersTests.java @@ -0,0 +1,316 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.ftp.inbound; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.ByteArrayInputStream; +import java.io.File; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.apache.commons.net.ftp.FTPFile; +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageChannels; +import org.springframework.integration.dsl.Pollers; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.DefaultSessionFactoryLocator; +import org.springframework.integration.file.remote.session.DelegatingSessionFactory; +import org.springframework.integration.file.remote.session.RotatingServerAdvice; +import org.springframework.integration.file.remote.session.RotatingServerAdvice.KeyDirectory; +import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.remote.session.SessionFactoryLocator; +import org.springframework.integration.file.remote.session.SessionFactoryMapBuilder; +import org.springframework.integration.ftp.FtpTestSupport; +import org.springframework.integration.ftp.dsl.Ftp; +import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; +import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; +import org.springframework.integration.metadata.SimpleMetadataStore; + +/** + * @author Gary Russell + * + * @since 5.0.7 + * + */ +public class RotatingServersTests extends FtpTestSupport { + + private static String tmpDir = localTemporaryFolder.getRoot().getAbsolutePath() + File.separator + "multiSF"; + + @BeforeClass + public static void setup() { + FtpRemoteFileTemplate rft = new FtpRemoteFileTemplate(sessionFactory()); + rft.execute(s -> { + s.mkdir("foo"); + s.mkdir("bar"); + s.mkdir("baz"); + s.mkdir("qux"); + s.mkdir("fiz"); + s.mkdir("buz"); + ByteArrayInputStream bais = new ByteArrayInputStream("foo".getBytes()); + s.write(bais, "foo/f1"); + s.write(bais, "baz/f2"); + s.write(bais, "fiz/f3"); + return null; + }); + } + + @Before + @After + public void clean() { + recursiveDelete(new File(tmpDir)); + } + + @Test + public void testStandard() throws Exception { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StandardConfig.class); + StandardConfig config = ctx.getBean(StandardConfig.class); + assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); + assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1); + File f1 = new File(tmpDir + File.separator + "standard" + File.separator + "f1"); + assertThat(f1.exists()).isTrue(); + File f2 = new File(tmpDir + File.separator + "standard" + File.separator + "f2"); + assertThat(f2.exists()).isTrue(); + File f3 = new File(tmpDir + File.separator + "standard" + File.separator + "f3"); + assertThat(f3.exists()).isTrue(); + assertThat(f1.delete()).isTrue(); + assertThat(f2.delete()).isTrue(); + assertThat(f3.delete()).isTrue(); + ctx.getBean("files", QueueChannel.class); + assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3); + ctx.close(); + } + + @Test + public void testFair() throws Exception { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(FairConfig.class); + StandardConfig config = ctx.getBean(StandardConfig.class); + assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); + assertThat(sfCalls).containsExactly(1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1, 2, 2, 3); + File f1 = new File(tmpDir + File.separator + "fair" + File.separator + "f1"); + assertThat(f1.exists()).isTrue(); + File f2 = new File(tmpDir + File.separator + "fair" + File.separator + "f2"); + assertThat(f2.exists()).isTrue(); + File f3 = new File(tmpDir + File.separator + "fair" + File.separator + "f3"); + assertThat(f3.exists()).isTrue(); + assertThat(f1.delete()).isTrue(); + assertThat(f2.delete()).isTrue(); + assertThat(f3.delete()).isTrue(); + assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3); + ctx.close(); + } + + @Test + public void testVariableLocalDir() throws Exception { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(VariableLocalConfig.class); + StandardConfig config = ctx.getBean(StandardConfig.class); + assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); + assertThat(sfCalls).containsExactly(1, 1, 1, 2, 2, 2, 3, 3, 3, 1, 1, 2, 2, 3, 3, 1, 1); + File f1 = new File(tmpDir + File.separator + "variable" + File.separator + "foo" + File.separator + "f1"); + assertThat(f1.exists()).isTrue(); + File f2 = new File(tmpDir + File.separator + "variable" + File.separator + "baz" + File.separator + "f2"); + assertThat(f2.exists()).isTrue(); + File f3 = new File(tmpDir + File.separator + "variable" + File.separator + "fiz" + File.separator + "f3"); + assertThat(f3.exists()).isTrue(); + assertThat(f1.delete()).isTrue(); + assertThat(f2.delete()).isTrue(); + assertThat(f3.delete()).isTrue(); + assertThat(f1.getParentFile().delete()).isTrue(); + assertThat(f2.getParentFile().delete()).isTrue(); + assertThat(f3.getParentFile().delete()).isTrue(); + assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3); + ctx.close(); + } + + @Test + public void testStreaming() throws Exception { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(StreamingConfig.class); + StandardConfig config = ctx.getBean(StandardConfig.class); + assertThat(config.latch.await(10, TimeUnit.SECONDS)).isTrue(); + List sfCalls = config.sessionSources.stream().limit(17).collect(Collectors.toList()); + // there's an extra getSession() with this adapter in listFiles + assertThat(sfCalls).containsExactly(1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 1, 1, 2, 2, 3); + ctx.getBean("files", QueueChannel.class); + assertThat(ctx.getBean("files", QueueChannel.class).getQueueSize()).isEqualTo(3); + ctx.close(); + } + + + @Configuration + @EnableIntegration + public static class StandardConfig { + + private final CountDownLatch latch = new CountDownLatch(17); + + List sessionSources = new ArrayList<>(); + + @Bean + public SessionFactory factory1() { + return new CachingSessionFactory(rawSessionFactory()) { + + @Override + public Session getSession() { + StandardConfig.this.sessionSources.add(1); + StandardConfig.this.latch.countDown(); + return super.getSession(); + } + + }; + } + + @Bean + public SessionFactory factory2() { + return new CachingSessionFactory(rawSessionFactory()) { + + @Override + public Session getSession() { + StandardConfig.this.sessionSources.add(2); + StandardConfig.this.latch.countDown(); + return super.getSession(); + } + + }; + } + + @Bean + public SessionFactory factory3() { + return new CachingSessionFactory(rawSessionFactory()) { + + @Override + public Session getSession() { + StandardConfig.this.sessionSources.add(3); + StandardConfig.this.latch.countDown(); + return super.getSession(); + } + + }; + } + + @Bean + public SessionFactoryLocator factoryLocator() { + return new DefaultSessionFactoryLocator<>(new SessionFactoryMapBuilder() + .put("one", factory1()) + .put("two", factory2()) + .put("three", factory3()) + .get()); + } + + @Bean + public DelegatingSessionFactory sf() { + return new DelegatingSessionFactory<>(factoryLocator()); + } + + @Bean + public RotatingServerAdvice advice() { + List keyDirectories = new ArrayList<>(); + keyDirectories.add(new KeyDirectory("one", "foo")); + keyDirectories.add(new KeyDirectory("one", "bar")); + keyDirectories.add(new KeyDirectory("two", "baz")); + keyDirectories.add(new KeyDirectory("two", "qux")); + keyDirectories.add(new KeyDirectory("three", "fiz")); + keyDirectories.add(new KeyDirectory("three", "buz")); + return theAdvice(keyDirectories); + } + + protected RotatingServerAdvice theAdvice(List keyDirectories) { + return new RotatingServerAdvice(sf(), keyDirectories); + } + + @Bean + public IntegrationFlow flow() { + return IntegrationFlows.from(Ftp.inboundAdapter(sf()) + .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate")) + .localDirectory(localDir()) + .remoteDirectory("."), + e -> e.poller(Pollers.fixedDelay(1).advice(advice()))) + .channel(MessageChannels.queue("files")) + .get(); + } + + protected File localDir() { + return new File(tmpDir, "standard"); + } + + } + + @Configuration + public static class FairConfig extends StandardConfig { + + @Override + protected RotatingServerAdvice theAdvice(List keyDirectories) { + return new RotatingServerAdvice(sf(), keyDirectories, true); + } + + @Override + protected File localDir() { + return new File(tmpDir, "fair"); + } + } + + @Configuration + public static class VariableLocalConfig extends StandardConfig { + + @Override + @Bean + public IntegrationFlow flow() { + return IntegrationFlows.from(Ftp.inboundAdapter(sf()) + .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate")) + .localDirectory(new File(tmpDir, "variable")) + .localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root") + .remoteDirectory("."), + e -> e.poller(Pollers.fixedDelay(1).advice(advice()))) + .channel(MessageChannels.queue("files")) + .get(); + } + + } + + @Configuration + public static class StreamingConfig extends StandardConfig { + + @Override + @Bean + public IntegrationFlow flow() { + return IntegrationFlows.from(Ftp.inboundStreamingAdapter(new FtpRemoteFileTemplate(sf())) + .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate")) + .remoteDirectory("."), + e -> e.poller(Pollers.fixedDelay(1).advice(advice()))) + .channel(MessageChannels.queue("files")) + .get(); + } + + } + +} diff --git a/spring-integration-ftp/src/test/resources/log4j2-test.xml b/spring-integration-ftp/src/test/resources/log4j2-test.xml index 7541cd10b7..5acad4f961 100644 --- a/spring-integration-ftp/src/test/resources/log4j2-test.xml +++ b/spring-integration-ftp/src/test/resources/log4j2-test.xml @@ -10,6 +10,7 @@ + diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index edf371a385..15ee2634f0 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -224,6 +224,8 @@ Convenience methods have been added so this can easily be done from a message fl IMPORTANT: When using session caching (see <>), each of the delegates should be cached; you cannot cache the `DelegatingSessionFactory` itself. +Starting with _version 5.0.7_, the `DelegatingSessionFactory` can be used in conjuction with a `RotatingServerAdvice` to poll multiple servers; see <>. + [[ftp-inbound]] === FTP Inbound Channel Adapter @@ -620,6 +622,91 @@ public class FtpJavaApplication { Notice that, in this example, the message handler downstream of the transformer has an advice that removes the remote file after processing. +[[ftp-rotating-server-advice]] +=== Inbound Channel Adapters: Polling Multiple Servers and Directories + +Starting with _version 5.0.7_, the `RotatingServerAdvice` is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories. +Configure the advice and add it to the poller's advice chain as normal. +A `DelegatingSessionFactory` is used to select the server see <> for more information. +The advice configuration consists of a list of `RotatingServerAdvice.KeyDirectory` objects. + +.Example +[source, java] +---- +@Bean +public RotatingServerAdvice advice() { + List keyDirectories = new ArrayList<>(); + keyDirectories.add(new KeyDirectory("one", "foo")); + keyDirectories.add(new KeyDirectory("one", "bar")); + keyDirectories.add(new KeyDirectory("two", "baz")); + keyDirectories.add(new KeyDirectory("two", "qux")); + keyDirectories.add(new KeyDirectory("three", "fiz")); + keyDirectories.add(new KeyDirectory("three", "buz")); + return new RotatingServerAdvice(delegatingSf(), keyDirectories); +} +---- + +This advice will poll directory `foo` on server `one` until no new files exist then move to directory `bar` and then directory `baz` on server `two`, etc. + +This default behavior can be modified with the `fair` constructor arg: + +.fair +[source, java] +---- +@Bean +public RotatingServerAdvice advice() { + ... + return new RotatingServerAdvice(delegatingSf(), keyDirectories, true); +} +---- + +In this case, the advice will move to the next server/directory regardless of whether the previous poll returned a file. + +Alternatively, you can provide your own `RotatingServerAdvice.RotationPolicy` to reconfigure the message source as needed: + +.policy +[source, java] +---- +public interface RotationPolicy { + + void beforeReceive(MessageSource source); + + void afterReceive(boolean messageReceived, MessageSource source); + +} +---- + +and + +.custom +[source, java] +---- +@Bean +public RotatingServerAdvice advice() { + return new RotatingServerAdvice(myRotationPolicy()); +} +---- + +The `local-filename-generator-expression` attribute (`localFilenameGeneratorExpression` on the synchronizer) can now contain the `#remoteDirectory` variable. +This allows files retrieved from different directories to be downloaded to similar directories locally: + +[source, java] +---- +@Bean +public IntegrationFlow flow() { + return IntegrationFlows.from(Ftp.inboundAdapter(sf()) + .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate")) + .localDirectory(new File(tmpDir)) + .localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root") + .remoteDirectory("."), + e -> e.poller(Pollers.fixedDelay(1).advice(advice()))) + .channel(MessageChannels.queue("files")) + .get(); +} +---- + +IMPORTANT: Do not configure a `TaskExecutor` on the poller when using this advice; see <> for more information. + [[ftp-max-fetch]] === Inbound Channel Adapters: Controlling Remote File Fetching diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index 3b28964284..e3dc2814a7 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -162,7 +162,15 @@ It enables you to examine and or reconfigure the source at this time. Returning Message afterReceive(Message result, MessageSource source) This method is called after the `receive()` method; again, you can reconfigure the source, or take any action perhaps depending on the result (which can be `null` if there was no message created by the source). -You can even return a different message! +You can even return a different message + +.Thread safety +[IMPORTANT] +==== +You should not configure the poller with a `TaskExecutor` if an advice mutates the `MessageSource`. +If an advice mutates the source, such mutations are not thread safe and could cause unexpected results, especially with high frequency pollers. +Consider using a downstream `ExecutorChannel` instead of adding an executor to the poller if you need to process poll results concurrently. +==== .Advice Chain Ordering [IMPORTANT] diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 98bc0611fb..47b803686b 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -239,6 +239,8 @@ Convenience methods have been added so this can easily be done from a message fl IMPORTANT: When using session caching (see <>), each of the delegates should be cached; you cannot cache the `DelegatingSessionFactory` itself. +Starting with _version 5.0.7_, the `DelegatingSessionFactory` can be used in conjuction with a `RotatingServerAdvice` to poll multiple servers; see <>. + [[sftp-session-caching]] === SFTP Session Caching @@ -665,6 +667,91 @@ public class SftpJavaApplication { Notice that, in this example, the message handler downstream of the transformer has an advice that removes the remote file after processing. +[[sftp-rotating-server-advice]] +=== Inbound Channel Adapters: Polling Multiple Servers and Directories + +Starting with _version 5.0.7_, the `RotatingServerAdvice` is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories. +Configure the advice and add it to the poller's advice chain as normal. +A `DelegatingSessionFactory` is used to select the server see <> for more information. +The advice configuration consists of a list of `RotatingServerAdvice.KeyDirectory` objects. + +.Example +[source, java] +---- +@Bean +public RotatingServerAdvice advice() { + List keyDirectories = new ArrayList<>(); + keyDirectories.add(new KeyDirectory("one", "foo")); + keyDirectories.add(new KeyDirectory("one", "bar")); + keyDirectories.add(new KeyDirectory("two", "baz")); + keyDirectories.add(new KeyDirectory("two", "qux")); + keyDirectories.add(new KeyDirectory("three", "fiz")); + keyDirectories.add(new KeyDirectory("three", "buz")); + return new RotatingServerAdvice(delegatingSf(), keyDirectories); +} +---- + +This advice will poll directory `foo` on server `one` until no new files exist then move to directory `bar` and then directory `baz` on server `two`, etc. + +This default behavior can be modified with the `fair` constructor arg: + +.fair +[source, java] +---- +@Bean +public RotatingServerAdvice advice() { + ... + return new RotatingServerAdvice(delegatingSf(), keyDirectories, true); +} +---- + +In this case, the advice will move to the next server/directory regardless of whether the previous poll returned a file. + +Alternatively, you can provide your own `RotatingServerAdvice.RotationPolicy` to reconfigure the message source as needed: + +.policy +[source, java] +---- +public interface RotationPolicy { + + void beforeReceive(MessageSource source); + + void afterReceive(boolean messageReceived, MessageSource source); + +} +---- + +and + +.custom +[source, java] +---- +@Bean +public RotatingServerAdvice advice() { + return new RotatingServerAdvice(myRotationPolicy()); +} +---- + +The `local-filename-generator-expression` attribute (`localFilenameGeneratorExpression` on the synchronizer) can now contain the `#remoteDirectory` variable. +This allows files retrieved from different directories to be downloaded to similar directories locally: + +[source, java] +---- +@Bean +public IntegrationFlow flow() { + return IntegrationFlows.from(Ftp.inboundAdapter(sf()) + .filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate")) + .localDirectory(new File(tmpDir)) + .localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root") + .remoteDirectory("."), + e -> e.poller(Pollers.fixedDelay(1).advice(advice()))) + .channel(MessageChannels.queue("files")) + .get(); +} +---- + +IMPORTANT: Do not configure a `TaskExecutor` on the poller when using this advice; see <> for more information. + [[sftp-max-fetch]] === Inbound Channel Adapters: Controlling Remote File Fetching diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e1f6f9c965..fe0a8b848c 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -202,6 +202,10 @@ New filters for detecting incomplete remote files are now provided. The `FtpOutboundGateway` and `SftpOutboundGateway` now support an option to remove the remote file after a successful transfer using the `GET` or `MGET` commands. +A `RotatingServerAdvice` is now available to poll multiple servers and/or directories with the inbound channel adapters. + +Also inbound adapter `localFilenameExpression` s can contain the variable `#remoteDirectory` which contains the remote directory being polled. + See <> and <> for more information. ==== Integration Properties