diff --git a/build.gradle b/build.gradle index f917e910c7..f839b2558c 100644 --- a/build.gradle +++ b/build.gradle @@ -161,7 +161,7 @@ subprojects { subproject -> testCompile project(":spring-integration-test-support") } - testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" + testCompile "org.slf4j:slf4j-log4j12:$slf4jVersion" } // enable all compiler warnings; individual projects may customize further diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/MessageSessionCallback.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/MessageSessionCallback.java index ed6aafe738..b4404bb78b 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/MessageSessionCallback.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/MessageSessionCallback.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -28,6 +28,7 @@ import org.springframework.messaging.Message; * @author Artem Bilan * @since 4.2 */ +@FunctionalInterface public interface MessageSessionCallback { /** diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/OperationsCallback.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/OperationsCallback.java new file mode 100644 index 0000000000..8977cc0ed8 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/OperationsCallback.java @@ -0,0 +1,44 @@ +/* + * Copyright 2017 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; + +/** + * Callback for using the same session for multiple + * RemoteFileTemplate operations. + + * @param the type the operations accepts. + * @param the type the callback returns. + * + * @author Artem Bilan + * + * @since 5.0 + */ +@FunctionalInterface +public interface OperationsCallback { + + /** + * Execute any number of operations using a dedicated remote + * session as long as those operations are performed + * on the template argument and on the calling thread. + * The session will be closed when the callback exits. + * + * @param operations the RemoteFileOperations. + * @return the result of operations. + */ + T doInOperations(RemoteFileOperations operations); + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java index a35dc7ee94..51c5971717 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java @@ -136,6 +136,16 @@ public interface RemoteFileOperations { */ T execute(SessionCallback callback); + /** + * Invoke the callback and run all operations on the template argument in a dedicated + * thread-bound session and reliably close the it afterwards. + * @param action the call back. + * @param the return type. + * @return the result from the {@link OperationsCallback#doInOperations(RemoteFileOperations)} + * @since 5.0 + */ + T invoke(OperationsCallback action); + /** * Execute the callback's doWithClient method after obtaining a session's * client, providing access to low level methods. diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index 8bf8afeb51..2d36ee788f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -23,6 +23,7 @@ import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -55,6 +56,7 @@ import org.springframework.util.StringUtils; * @author David Turanski * @author Gary Russell * @author Artem Bilan + * * @since 3.0 * */ @@ -67,6 +69,13 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ */ protected final SessionFactory sessionFactory; + /* + * Not static as normal since we want this TL to be scoped within the template instance. + */ + private final ThreadLocal> contextSessions = new ThreadLocal<>(); + + private final AtomicInteger activeTemplateCallbacks = new AtomicInteger(); + private volatile String temporaryFileSuffix = ".writing"; private volatile boolean autoCreateDirectory = false; @@ -398,6 +407,14 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ @Override public Session getSession() { + if (this.activeTemplateCallbacks.get() > 0) { + Session session = this.contextSessions.get(); + // If no session in the ThreadLocal, no {@code invoke()} in this call stack + if (session != null) { + return session; + } + } + return this.sessionFactory.getSession(); } @@ -405,9 +422,17 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ @Override public T execute(SessionCallback callback) { Session session = null; + boolean invokeScope = false; + if (this.activeTemplateCallbacks.get() > 0) { + session = this.contextSessions.get(); + } try { - session = this.sessionFactory.getSession(); - Assert.notNull(session, "failed to acquire a Session"); + if (session == null) { + session = this.sessionFactory.getSession(); + } + else { + invokeScope = true; + } return callback.doInSession(session); } catch (Exception e) { @@ -420,7 +445,7 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ throw new MessagingException("Failed to execute on session", e); } finally { - if (session != null) { + if (!invokeScope) { try { session.close(); } @@ -433,6 +458,29 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializ } } + @Override + public T invoke(OperationsCallback action) { + Session contextSession = this.contextSessions.get(); + if (contextSession == null) { + this.contextSessions.set(this.sessionFactory.getSession()); + } + this.activeTemplateCallbacks.incrementAndGet(); + + try { + return action.doInOperations(this); + } + finally { + this.activeTemplateCallbacks.decrementAndGet(); + if (contextSession == null) { + Session session = this.contextSessions.get(); + if (session != null) { + session.close(); + } + this.contextSessions.remove(); + } + } + } + @Override public T executeWithClient(ClientCallback callback) { throw new UnsupportedOperationException("executeWithClient() is not supported by the generic template"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java index 036a144f3e..d463bf83cf 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java @@ -24,6 +24,9 @@ import org.springframework.integration.file.remote.session.Session; * Callback invoked by {@code RemoteFileOperations.execute()} - allows multiple operations * on a session. * + * @param the type the operations accepts. + * @param the type the callback returns. + * * @author Gary Russell * @since 3.0 * 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 ede918b499..7c072fa6ad 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 @@ -42,12 +42,15 @@ import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.AbstractFileInfo; import org.springframework.integration.file.remote.MessageSessionCallback; +import org.springframework.integration.file.remote.RemoteFileOperations; import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.RemoteFileUtils; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.support.FileExistsMode; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; +import org.springframework.integration.support.MutableMessage; import org.springframework.integration.support.PartialSuccessException; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -80,6 +83,11 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply */ LS("ls"), + /** + * List remote file names. + */ + NLST("nlst"), + /** * Retrieve a remote file. */ @@ -551,6 +559,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply switch (this.command) { case LS: return doLs(requestMessage); + case NLST: + return doNlst(requestMessage); case GET: return doGet(requestMessage); case MGET: @@ -575,13 +585,45 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply dir += this.remoteFileTemplate.getRemoteFileSeparator(); } final String fullDir = dir; - List payload = this.remoteFileTemplate.execute(session -> - AbstractRemoteFileOutboundGateway.this.ls(session, fullDir)); + List payload = this.remoteFileTemplate.execute(session -> ls(requestMessage, session, fullDir)); return getMessageBuilderFactory() .withPayload(payload) .setHeader(FileHeaders.REMOTE_DIRECTORY, dir); } + private Object doNlst(Message requestMessage) { + String dir = this.fileNameProcessor.processMessage(requestMessage); + if (dir != null && !dir.endsWith(this.remoteFileTemplate.getRemoteFileSeparator())) { + 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); + } + + /** + * List remote files names for the provided directory. + * The message can be consulted for some context related to the current request; + * isn't used in the default implementation. + * @param message the message related to the current request + * @param session the session to perform list file names command + * @param dir the remote directory to list file names + * @return the list of file/directory names in the provided dir + * @throws IOException the IO exception during performing remote command + * @since 5.0 + */ + protected List nlst(Message message, Session session, String dir) throws IOException { + String remoteDirectory = buildRemotePath(dir, ""); + List fileNames = Arrays.asList(session.listNames(remoteDirectory)); + if (!this.options.contains(Option.NOSORT)) { + Collections.sort(fileNames); + } + return fileNames; + } + private Object doGet(final Message requestMessage) { final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage); final String remoteFilename = getRemoteFilename(remoteFilePath); @@ -627,7 +669,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply String remoteFilename = getRemoteFilename(remoteFilePath); String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename); - boolean payload = this.remoteFileTemplate.remove(remoteFilePath); + boolean payload = this.remoteFileTemplate.execute(session -> rm(requestMessage, session, remoteFilePath)); return getMessageBuilderFactory() .withPayload(payload) @@ -635,6 +677,21 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply .setHeader(FileHeaders.REMOTE_FILE, remoteFilename); } + /** + * Perform remote delete for the provided path. + * The message can be consulted to determine some context; + * isn't used in the default implementation. + * @param message the request message related to the path to remove + * @param session the remote protocol session to perform remove command + * @param remoteFilePath the remote path to remove + * @return true or false as a result of the remote removal + * @throws IOException the IO exception during performing remote command + * @since 5.0 + */ + protected boolean rm(Message message, Session session, String remoteFilePath) throws IOException { + return session.remove(remoteFilePath); + } + private Object doMv(Message requestMessage) { String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage); String remoteFilename = getRemoteFilename(remoteFilePath); @@ -642,38 +699,80 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply String remoteFileNewPath = this.renameProcessor.processMessage(requestMessage); Assert.hasLength(remoteFileNewPath, "New filename cannot be empty"); - this.remoteFileTemplate.rename(remoteFilePath, remoteFileNewPath); + Boolean result = + this.remoteFileTemplate.execute(session -> + mv(requestMessage, session, remoteFilePath, remoteFileNewPath)); + return getMessageBuilderFactory() - .withPayload(Boolean.TRUE) + .withPayload(result) .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) .setHeader(FileHeaders.REMOTE_FILE, remoteFilename) .setHeader(FileHeaders.RENAME_TO, remoteFileNewPath); } + /** + * Move one remote path to another. + * The message can be consulted to determine some context; + * isn't used in the default implementation. + * @param message the request message related to this move command + * @param session the remote protocol session to perform move command + * @param remoteFilePath the source remote path + * @param remoteFileNewPath the target remote path + * @return true or false as a result of the operation + * @throws IOException the IO exception during performing remote command + * @since 5.0 + */ + protected boolean mv(Message message, Session session, String remoteFilePath, String remoteFileNewPath) + throws IOException { + int lastSeparator = remoteFileNewPath.lastIndexOf(this.remoteFileTemplate.getRemoteFileSeparator()); + if (lastSeparator > 0) { + String remoteFileDirectory = remoteFileNewPath.substring(0, lastSeparator + 1); + RemoteFileUtils.makeDirectories(remoteFileDirectory, session, + this.remoteFileTemplate.getRemoteFileSeparator(), this.logger); + } + session.rename(remoteFilePath, remoteFileNewPath); + return true; + } + private String doPut(Message requestMessage) { return doPut(requestMessage, null); } private String doPut(Message requestMessage, String subDirectory) { - String path = this.remoteFileTemplate.send(requestMessage, subDirectory, this.fileExistsMode); + return this.remoteFileTemplate.invoke(template -> + put(requestMessage, template.getSession(), subDirectory)); + } + + /** + * Put the file based on the message to the remote server. + * The message can be consulted to determine some context. + * The session argument isn't used in the default implementation. + * @param message the request message related to this put command + * @param session the remote protocol session related to this invocation context + * @param subDirectory the target sub directory to put + * @since 5.0 + */ + protected String put(Message message, Session session, String subDirectory) { + String path = this.remoteFileTemplate.send(message, subDirectory, this.fileExistsMode); if (path == null) { - throw new MessagingException(requestMessage, "No local file found for " + requestMessage); + throw new MessagingException(message, "No local file found for " + message); } if (this.chmod != null && isChmodCapable()) { doChmod(this.remoteFileTemplate, path, this.chmod); } return path; + } /** * Set the mode on the remote file after transfer; the default implementation does * nothing. - * @param remoteFileTemplate the remote file template. + * @param remoteFileOperations the remote file template. * @param path the path. * @param chmod the chmod to set. * @since 4.3 */ - protected void doChmod(RemoteFileTemplate remoteFileTemplate, String path, int chmod) { + protected void doChmod(RemoteFileOperations remoteFileOperations, String path, int chmod) { // no-op } @@ -689,26 +788,40 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply throw new IllegalArgumentException("Only File or String payloads allowed for 'mput'"); } if (!file.isDirectory()) { - return this.doPut(requestMessage); + return doPut(requestMessage); } else { - return putLocalDirectory(requestMessage, file, null); + File localDir = file; + return this.remoteFileTemplate.invoke(t -> + mPut(requestMessage, t.getSession(), localDir)); } } + /** + * Put files from the provided directory to the remote server recursively. + * The message can be consulted to determine some context. + * The session argument isn't used in the default implementation. + * @param message the request message related to this mPut command + * @param session the remote protocol session for this invocation context + * @param localDir the local directory to mput to the server + * @since 5.0 + */ + protected List mPut(Message message, Session session, File localDir) { + return putLocalDirectory(message, localDir, null); + } + private List putLocalDirectory(Message requestMessage, File file, String subDirectory) { File[] files = file.listFiles(); List filteredFiles = this.filterMputFiles(files); - List replies = new ArrayList(); + List replies = new ArrayList<>(); try { for (File filteredFile : filteredFiles) { if (!filteredFile.isDirectory()) { - String path = this.doPut(this.getMessageBuilderFactory().withPayload(filteredFile) - .copyHeaders(requestMessage.getHeaders()) - .build(), subDirectory); + String path = doPut(new MutableMessage<>(filteredFile, requestMessage.getHeaders()), subDirectory); if (path == null) { //NOSONAR - false positive if (logger.isDebugEnabled()) { - logger.debug("File " + filteredFile.getAbsolutePath() + " removed before transfer; ignoring"); + logger.debug("File " + filteredFile.getAbsolutePath() + + " removed before transfer; ignoring"); } } else { @@ -719,7 +832,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply String newSubDirectory = (StringUtils.hasText(subDirectory) ? subDirectory + this.remoteFileTemplate.getRemoteFileSeparator() : "") + filteredFile.getName(); - replies.addAll(this.putLocalDirectory(requestMessage, filteredFile, newSubDirectory)); + replies.addAll(putLocalDirectory(requestMessage, filteredFile, newSubDirectory)); } } } @@ -734,14 +847,24 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply "Partially successful 'mput' operation" + (subDirectory == null ? "" : (" on " + subDirectory)), e, replies, filteredFiles); } - else if (e instanceof MessagingException) { - throw (MessagingException) e; + else { + throw e; } } return replies; } - protected List ls(Session session, String dir) throws IOException { + /** + * List remote files to local representation. + * The message can be consulted for some context for the current request; + * isn't used in the default implementation. + * @param message the message related to the list request + * @param session the session to perform list command + * @param dir the remote directory to list content + * @return the list of remote files + * @throws IOException the IO exception during performing remote command + */ + protected List ls(Message message, Session session, String dir) throws IOException { List lsFiles = listFilesInRemoteDir(session, dir, ""); if (!this.options.contains(Option.LINKS)) { purgeLinks(lsFiles); @@ -842,8 +965,6 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply /** * Copy a remote file to the configured local directory. - * - * * @param message the message. * @param session the session. * @param remoteDir the remote directory. @@ -955,7 +1076,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply List files = new ArrayList(); String remotePath = buildRemotePath(remoteDirectory, remoteFilename); @SuppressWarnings("unchecked") - List> remoteFiles = (List>) ls(session, remotePath); + List> remoteFiles = (List>) ls(message, session, remotePath); if (remoteFiles.size() == 0 && this.options.contains(Option.EXCEPTION_WHEN_EMPTY)) { throw new MessagingException("No files found at " + (remoteDirectory != null ? remoteDirectory : "Client Working Directory") @@ -1001,7 +1122,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply String remoteFilename) throws IOException { List files = new ArrayList(); @SuppressWarnings("unchecked") - List> fileNames = (List>) ls(session, remoteDirectory); + List> fileNames = (List>) ls(message, session, remoteDirectory); if (fileNames.size() == 0 && this.options.contains(Option.EXCEPTION_WHEN_EMPTY)) { throw new MessagingException("No files found at " + (remoteDirectory != null ? remoteDirectory : "Client Working Directory") diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd index e9caee535e..154e533020 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd @@ -891,6 +891,7 @@ Only files matching this regular expression will be picked up by this adapter. + diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParser.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParser.java index 12826279a6..d404f46285 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParser.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -20,6 +20,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.file.config.AbstractRemoteFileOutboundGatewayParser; import org.springframework.integration.file.remote.RemoteFileOperations; import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; @@ -30,6 +31,7 @@ import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; /** * @author Gary Russell * @author Artem Bilan + * * @since 2.1 * */ @@ -66,6 +68,9 @@ public class FtpOutboundGatewayParser extends AbstractRemoteFileOutboundGatewayP .getValue(); templateDefinition.getPropertyValues() .add("existsMode", FtpRemoteFileTemplate.ExistsMode.NLST); + + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "working-dir-expression", + "workingDirExpressionString"); } } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpOutboundGatewaySpec.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpOutboundGatewaySpec.java index 9b3426246f..ed3d7e92f6 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpOutboundGatewaySpec.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpOutboundGatewaySpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -16,22 +16,28 @@ package org.springframework.integration.ftp.dsl; +import java.util.function.Function; + import org.apache.commons.net.ftp.FTPFile; +import org.springframework.expression.Expression; +import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.file.dsl.RemoteFileOutboundGatewaySpec; -import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; +import org.springframework.integration.ftp.gateway.FtpOutboundGateway; +import org.springframework.messaging.Message; /** * A {@link RemoteFileOutboundGatewaySpec} for FTP. * * @author Artem Bilan + * * @since 5.0 */ public class FtpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec { - FtpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway outboundGateway) { + FtpOutboundGatewaySpec(FtpOutboundGateway outboundGateway) { super(outboundGateway); } @@ -51,4 +57,36 @@ public class FtpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec, String> workingDirFunction) { + ((FtpOutboundGateway) this.target).setWorkingDirExpression(new FunctionExpression<>(workingDirFunction)); + return this; + } + + + } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java index 7275f64b63..bf7e67583f 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/gateway/FtpOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -16,29 +16,45 @@ package org.springframework.integration.ftp.gateway; +import java.io.File; +import java.io.IOException; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Objects; +import java.util.concurrent.Callable; +import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.file.remote.AbstractFileInfo; import org.springframework.integration.file.remote.MessageSessionCallback; import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.session.FtpFileInfo; import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandlingException; /** * Outbound Gateway for performing remote file operations via FTP/FTPS. * * @author Gary Russell * @author Artem Bilan + * * @since 2.1 */ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway { + private Expression workingDirExpression; + + private StandardEvaluationContext evaluationContext; + /** * Construct an instance using the provided session factory and callback for * performing operations on the session. @@ -111,6 +127,26 @@ public class FtpOutboundGateway extends AbstractRemoteFileOutboundGateway ls(Message message, Session session, String dir) throws IOException { + return doInWorkingDirectory(message, session, + () -> super.ls(message, session, dir)); + } + + @Override + protected List nlst(Message message, Session session, String dir) throws IOException { + return doInWorkingDirectory(message, session, + () -> super.nlst(message, session, dir)); + } + + @Override + protected File get(Message message, Session session, String remoteDir, String remoteFilePath, + String remoteFilename, FTPFile fileInfoParam) throws IOException { + return doInWorkingDirectory(message, session, + () -> super.get(message, session, remoteDir, remoteFilePath, remoteFilename, fileInfoParam)); + } + + @Override + protected List mGet(Message message, Session session, String remoteDirectory, + String remoteFilename) throws IOException { + return doInWorkingDirectory(message, session, + () -> super.mGet(message, session, remoteDirectory, remoteFilename)); + } + + @Override + protected boolean rm(Message message, Session session, String remoteFilePath) throws IOException { + return doInWorkingDirectory(message, session, + () -> super.rm(message, session, remoteFilePath)); + } + + @Override + protected boolean mv(Message message, Session session, String remoteFilePath, String remoteFileNewPath) + throws IOException { + return doInWorkingDirectory(message, session, + () -> super.mv(message, session, remoteFilePath, remoteFileNewPath)); + } + + @Override + protected String put(Message message, Session session, String subDirectory) { + try { + return doInWorkingDirectory(message, session, + () -> super.put(message, session, subDirectory)); + } + catch (IOException e) { + throw new MessageHandlingException(message, "Cannot handle PUT command", e); + } + } + + @Override + protected List mPut(Message message, Session session, File localDir) { + try { + return doInWorkingDirectory(message, session, + () -> super.mPut(message, session, localDir)); + } + catch (IOException e) { + throw new MessageHandlingException(message, "Cannot handle MPUT command", e); + } + } + + private V doInWorkingDirectory(Message message, Session session, Callable task) + throws IOException { + Expression workingDirExpression = this.workingDirExpression; + FTPClient ftpClient = (FTPClient) session.getClientInstance(); + String currentWorkingDirectory = null; + boolean restoreWorkingDirectory = false; + try { + if (workingDirExpression != null) { + currentWorkingDirectory = ftpClient.printWorkingDirectory(); + String newWorkingDirectory = + workingDirExpression.getValue(this.evaluationContext, message, String.class); + if (!Objects.equals(currentWorkingDirectory, newWorkingDirectory)) { + ftpClient.changeWorkingDirectory(newWorkingDirectory); + restoreWorkingDirectory = true; + } + } + return task.call(); + } + catch (Exception e) { + if (e instanceof IOException) { + throw (IOException) e; + + } + else if (e instanceof RuntimeException) { + throw (RuntimeException) e; + } + else { + throw new IOException("Uncategorised IO exception", e); + } + } + finally { + if (restoreWorkingDirectory) { + ftpClient.changeWorkingDirectory(currentWorkingDirectory); + } + } + } + } diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd index bbb61dbcb7..d27110246f 100644 --- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd +++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd @@ -220,6 +220,13 @@ + + + + The SpEL expression to evaluate FTP client working directory against request message. + + + diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml index c323d191e5..67a5000b2f 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests-context.xml @@ -182,6 +182,12 @@ command-options="-1" reply-channel="output"/> + + ("foo")); + Message receive = this.output.receive(10000); + assertNotNull(receive); + assertThat(receive.getPayload(), instanceOf(List.class)); + List files = (List) receive.getPayload(); + assertEquals(3, files.size()); + assertThat(files, containsInAnyOrder("subFtpSource", " ftpSource1.txt", "ftpSource2.txt")); + + FTPFile[] ftpFiles = ftpSessionFactory.getSession().list(null); + for (FTPFile ftpFile : ftpFiles) { + if (!ftpFile.isDirectory()) { + assertTrue(files.contains(ftpFile.getName())); + } + } + } + @Test public void testInboundChannelAdapterWithNullDir() throws IOException { Session session = ftpSessionFactory.getSession(); diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java index 1e4aa4aa80..7fff432cd4 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/gateway/SftpOutboundGateway.java @@ -24,6 +24,7 @@ import java.util.List; import org.springframework.integration.file.remote.AbstractFileInfo; import org.springframework.integration.file.remote.ClientCallbackWithoutResult; import org.springframework.integration.file.remote.MessageSessionCallback; +import org.springframework.integration.file.remote.RemoteFileOperations; import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; import org.springframework.integration.file.remote.session.SessionFactory; @@ -147,8 +148,8 @@ public class SftpOutboundGateway extends AbstractRemoteFileOutboundGateway remoteFileTemplate, final String path, final int chmod) { - remoteFileTemplate.executeWithClient((ClientCallbackWithoutResult) client -> { + protected void doChmod(RemoteFileOperations remoteFileOperations, final String path, final int chmod) { + remoteFileOperations.executeWithClient((ClientCallbackWithoutResult) client -> { try { client.chmod(chmod, path); } diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 4c7190301c..b81ae9f27e 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -768,6 +768,7 @@ The _FTP Outbound Gateway_ provides a limited set of commands to interact with a Commands supported are: * ls (list files) +* nlst (list file names) * get (retrieve file) * mget (retrieve file(s)) * rm (remove file(s)) @@ -804,6 +805,22 @@ From Java perspective there are two new constructor without `expression` argumen The `null` for `LS` command is treated as an Client working directory according to the FTP protocol. The working directory can be set via the `FTPClient.changeWorkingDirectory()` function when you extend the `DefaultFtpSessionFactory` and implement `postProcessClientAfterConnect()` callback. +*nlst* + +(Since _version 5.0_) + +Lists remote file names and supports the following options: + +* -f - do not sort the list + +The message payload resulting from an _nlst_ operation is a list of file names. + +The remote directory that the _nlst_ command acted on is provided in the `file_remoteDirectory` header. + +Unlike the `-1` option for the _ls_ command (see above), which uses the `LIST` command, the _nlst_ command sends an `NLST` command to the target FTP server. +This command is useful when the server doesn't support `LIST`, due to security restrictions, for example. +The result of the _nlst_ is just the names, therefore the framework can't determine if an entity is a directory, to perform filtering or recursive listing, for example. + *get* _get_ retrieves a remote file and supports the following option: @@ -986,11 +1003,9 @@ Here is an example of a gateway configured for an ls command... reply-channel="toSplitter"/> ---- -The payload of the message sent to the toSplitter channel is a list of String objects containing the filename of each -file. +The payload of the message sent to the `toSplitter` channel is a list of String objects containing the filename of each file. If the `command-options` was omitted, it would be a list of `FileInfo` objects. -Options are provided space-delimited, e.g. -`command-options="-1 -dirs -links"`. +Options are provided space-delimited, e.g. `command-options="-1 -dirs -links"`. Starting with _version 4.2_, the `GET`, `MGET`, `PUT` and `MPUT` commands support a `FileExistsMode` property (`mode` when using the namespace support). This affects the behavior when the local file exists (`GET` and `MGET`) or the remote @@ -998,6 +1013,9 @@ file exists (`PUT` and `MPUT`). Supported modes are `REPLACE`, `APPEND`, `FAIL` For backwards compatibility, the default mode for `PUT` and `MPUT` operations is `REPLACE` and for `GET` and `MGET` operations, the default is `FAIL`. +Starting with _version 5.0_, the `setWorkingDirExpression()` (`working-dir-expression`) option is provided on the `FtpOutboundGateway` (``) enabling the client working directory to be changed at runtime; the expression is evaluated against the request message. +The previous working directory is restored after each gateway operation. + ==== Configuring with Java Configuration The following Spring Boot application provides an example of configuring the Outbound Gateway using Java configuration: @@ -1179,6 +1197,12 @@ Since we know that the `FileExistsMode.FAIL` case is always only looking for a f For any other cases the `FtpRemoteFileTemplate` can be extended for implementing a custom logic in the overridden `exist()` method. +Starting with _version 5.0_, the new `RemoteFileOperations.invoke(OperationsCallback action)` method is available. +This method allows several `RemoteFileOperations` calls to be called in the scope of the same, thread-bounded, `Session`. +This is useful when you need to perform several high-level operations of the `RemoteFileTemplate` as one unit of work. +For example `AbstractRemoteFileOutboundGateway` uses it with the _mput_ command implementation, where we perform a _put_ operation for each file in the provided directory and recursively for its sub-directories. +See the JavaDocs for more information. + [[ftp-session-callback]] === MessageSessionCallback diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 2b4afa4c8c..c3a03378ef 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -289,6 +289,13 @@ For more information, refer to the http://docs.spring.io/spring-integration/api/ Additional methods were added in _version 4.1_ including `getClientInstance()` which provides access to the underlying `ChannelSftp` enabling access to low-level APIs. +Starting with _version 5.0_, the new `RemoteFileOperations.invoke(OperationsCallback action)` method is available. +This method allows several `RemoteFileOperations` calls to be called in the scope of the same, thread-bounded, `Session`. +This is useful when you need to perform several high-level operations of the `RemoteFileTemplate` as one unit of work. +For example `AbstractRemoteFileOutboundGateway` uses it with the _mput_ command implementation, where we perform a _put_ operation for each file in the provided directory and recursively for its sub-directories. +See the JavaDocs for more information. + + [[sftp-inbound]] === SFTP Inbound Channel Adapter @@ -791,6 +798,7 @@ The _SFTP Outbound Gateway_ provides a limited set of commands to interact with Commands supported are: * ls (list files) +* nlst (list file names) * get (retrieve file) * mget (retrieve file(s)) * rm (remove file(s)) @@ -821,6 +829,20 @@ If the `-dirs` option is included, each recursive directory is also returned as In this case, it is recommended that the `-1` is not used because you would not be able to determine files Vs. directories, which is achievable using the `FileInfo` objects. +*nlst* + +(Since _version 5.0_) + +Lists remote file names and supports the following options: + +* -f - do not sort the list + +The message payload resulting from an _nlst_ operation is a list of file names. + +The remote directory that the _nlst_ command acted on is provided in the `file_remoteDirectory` header. + +The SFTP protocol doesn't provide _list names_ functionality, s this command is fully equivalent of the _ls_ command with `-1` option and added here for convenience. + *get* _get_ retrieves a remote file and supports the following option: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 1d80affba6..917d6c0527 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -136,6 +136,12 @@ The FTP and SFTP outbound channel adapters, as well as `PUT` command of the outb The inbound channel adapters now can build file tree locally and use a new `RecursiveDirectoryScanner` by default for local directory. Also these adapters can now be switched to the `WatchService` instead. +The `NLST` command has been added to the `AbstractRemoteFileOutboundGateway` to perform only list files names remote command. + +The `FtpOutboundGateway` can now be supplied with `workingDirExpression` to change the FTP client working directory for the current request message. + +The `RemoteFileTemplate` is supplied now with the `invoke(OperationsCallback action)` to perform several `RemoteFileOperations` calls in the scope of the same, thread-bounded, `Session`. + See <> and <> for more information.