diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java index d3d6bec26f..552ee1b6e5 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java @@ -178,7 +178,7 @@ public class FileTransferringMessageHandler extends AbstractMessageHandler { remoteDirectory += File.separatorChar; } String remoteFilePath = remoteDirectory + file.getName(); - session.put(fileInputStream, remoteFilePath); + session.copy(fileInputStream, remoteFilePath); fileInputStream.close(); return true; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index 77429da910..3a1fbbffb6 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -16,7 +16,9 @@ package org.springframework.integration.file.remote.session; +import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.locks.ReentrantLock; @@ -116,20 +118,20 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { } } - public boolean rm(String path) { - return this.targetSession.rm(path); + public boolean remove(String path) { + return this.targetSession.remove(path); } - public F[] ls(String path) { - return this.targetSession.ls(path); + public F[] list(String path) { + return this.targetSession.list(path); } - public InputStream get(String source) { - return this.targetSession.get(source); + public void copy(String source, OutputStream os) throws IOException{ + this.targetSession.copy(source, os); } - public void put(InputStream inputStream, String destination) { - this.targetSession.put(inputStream, destination); + public void copy(InputStream inputStream, String destination) throws IOException{ + this.targetSession.copy(inputStream, destination); } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java index 0adf602bd4..2e7dd839e1 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java @@ -16,7 +16,9 @@ package org.springframework.integration.file.remote.session; +import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; /** * Common abstraction for a Session with a remote File system. @@ -24,18 +26,18 @@ import java.io.InputStream; * @author Josh Long * @author Mario Gray * @author Mark Fisher + * @author Oleg Zhurakousky * @since 2.0 */ public interface Session { - boolean rm(String path); + boolean remove(String path); - F[] ls(String path); + F[] list(String path); + + void copy(String source, OutputStream outputStream) throws IOException; - InputStream get(String source); - - void put(InputStream inputStream, String destination); + void copy(InputStream inputStream, String destination) throws IOException; void close(); - } 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 7628b322e4..1d6dfebd7f 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 @@ -33,7 +33,6 @@ import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.util.Assert; -import org.springframework.util.FileCopyUtils; import org.springframework.util.ObjectUtils; /** @@ -117,7 +116,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS try { session = this.sessionFactory.getSession(); Assert.state(session != null, "failed to acquire a Session"); - F[] files = session.ls(this.remoteDirectory); + F[] files = session.list(this.remoteDirectory); if (!ObjectUtils.isEmpty(files)) { Collection filteredFiles = this.filterFiles(files); for (F file : filteredFiles) { @@ -160,10 +159,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS InputStream inputStream = null; FileOutputStream fileOutputStream = new FileOutputStream(tempFile); try { - inputStream = session.get(remoteFilePath); - if (inputStream != null) { - FileCopyUtils.copy(inputStream, fileOutputStream); - } + session.copy(remoteFilePath, fileOutputStream); } catch (Exception e) { if (e instanceof RuntimeException){ @@ -189,7 +185,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS } if (tempFile.renameTo(localFile)) { if (this.deleteRemoteFiles) { - session.rm(remoteFilePath); + session.remove(remoteFilePath); if (logger.isDebugEnabled()) { logger.debug("deleted " + remoteFilePath); } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java index d54a974cfd..a7ed22e85e 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java @@ -13,12 +13,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.integration.ftp.session; import java.io.IOException; import java.net.SocketException; +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTP; @@ -26,6 +27,7 @@ import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPClientConfig; import org.apache.commons.net.ftp.FTPReply; +import org.springframework.aop.framework.ProxyFactory; import org.springframework.integration.MessagingException; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; @@ -36,13 +38,13 @@ import org.springframework.util.Assert; * * @author Iwein Fuld * @author Mark Fisher + * @author Oleg Zhurakousky * @since 2.0 */ public abstract class AbstractFtpSessionFactory implements SessionFactory { public static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/"; - private final Log logger = LogFactory.getLog(this.getClass()); protected FTPClientConfig config; @@ -54,8 +56,8 @@ public abstract class AbstractFtpSessionFactory implements protected String password; protected int port = FTP.DEFAULT_PORT; - - protected String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY; + + protected int bufferSize = 2048; //see https://issues.apache.org/jira/browse/NET-207 protected int clientMode = FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE; @@ -80,6 +82,10 @@ public abstract class AbstractFtpSessionFactory implements Assert.notNull(config); this.config = config; } + + public void setBufferSize(int bufferSize) { + this.bufferSize = bufferSize; + } public void setHost(String host) { Assert.hasText(host); @@ -100,12 +106,6 @@ public abstract class AbstractFtpSessionFactory implements Assert.notNull(pass, "password should not be null"); this.password = pass; } - - public void setRemoteWorkingDirectory(String remoteWorkingDirectory) { - Assert.notNull(remoteWorkingDirectory, "remote directory should not be null"); - this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/"); - } - /** * ACTIVE_LOCAL_DATA_CONNECTION_MODE = 0
* A constant indicating the FTP session is expecting all transfers @@ -139,41 +139,46 @@ public abstract class AbstractFtpSessionFactory implements } } + @SuppressWarnings("unchecked") private T createClient() throws SocketException, IOException { - T client = this.createClientInstance(); + final T client = this.createClientInstance(); Assert.notNull(client, "client must not be null"); client.configure(this.config); Assert.hasText(this.username, "username is required"); - client.connect(this.host); - this.afterConnect(client); - if (!FTPReply.isPositiveCompletion(client.getReplyCode())) { - throw new MessagingException("Connecting to server [" + - this.host + ":" + this.port + "] failed. Please check the connection."); - } - if (logger.isDebugEnabled()) { - logger.debug("Connected to server [" + this.host + ":" + this.port + "]"); - } - if (!client.login(username, password)) { - throw new MessagingException( - "Login failed. Please check the username and password."); - } - this.updateClientMode(client); - client.setFileType(this.fileType); - if (logger.isDebugEnabled()) { - logger.debug("login successful"); - } - if (!this.remoteWorkingDirectory.equals(client.printWorkingDirectory()) && - !client.changeWorkingDirectory(this.remoteWorkingDirectory)) { - throw new MessagingException("Could not change directory to '" + - remoteWorkingDirectory + "'. Please check the path."); - } - if (logger.isDebugEnabled()) { - logger.debug("working directory is: " + client.printWorkingDirectory()); - } - if (client != null) { - this.postProcessClient(client); - } - return client; + + this.postProcessClientBeforeConnect(client); + + ProxyFactory factory = new ProxyFactory(client); + factory.setProxyTargetClass(true); + factory.addAdvice(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + String methodName = invocation.getMethod().getName(); + if (!methodName.endsWith("connect")){ // will take care of both 'connect' and 'disconnect' + if (!client.isConnected()){ + client.connect(host); + postProcessClientAfterConnect (client); + if (!FTPReply.isPositiveCompletion(client.getReplyCode())) { + throw new MessagingException("Connecting to server [" + + host + ":" + port + "] failed. Please check the connection."); + } + + logger.debug("Connected to server [" + host + ":" + port + "]"); + + if (!client.login(username, password)) { + throw new IllegalStateException("Login failed. The respponse from the server is: " + + client.getReplyString()); + } + + updateClientMode(client); + client.setFileType(fileType); + client.setBufferSize(bufferSize); + } + } + return invocation.proceed(); + } + }); + + return (T) factory.getProxy(); } /** @@ -195,18 +200,16 @@ public abstract class AbstractFtpSessionFactory implements protected abstract T createClientInstance(); /** - * this is a hook to setup the state of the {@link org.apache.commons.net.ftp.FTPClient} impl *after* the - * implementation's {@link org.apache.commons.net.ftp.FTPClient#connect(String)} method's been called but before any - * action's been taken. - * - * @param t the ftp client instance on which to act - * @throws IOException if anything should go wrong + * Will handle additional initialization after client.connect() method was invoked, + * but before any action on the client has been taken */ - protected void afterConnect(T t) throws IOException { + protected void postProcessClientAfterConnect(T t) throws IOException { // NOOP } - - protected void postProcessClient(T t) throws IOException { + /** + * Will handle additional initialization before client.connect() method was invoked. + */ + protected void postProcessClientBeforeConnect(T client) throws IOException { // NOOP } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/DefaultFtpsSessionFactory.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/DefaultFtpsSessionFactory.java index 044433fab3..12e9ed7b2d 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/DefaultFtpsSessionFactory.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/DefaultFtpsSessionFactory.java @@ -123,13 +123,13 @@ public class DefaultFtpsSessionFactory extends AbstractFtpSessionFactory - - - - + - - - + - @@ -19,8 +18,5 @@ - - - + channel="ftpChannel"/> diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml index 15e3d723de..1f48dcda1f 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsInboundChannelAdapterParserTests-context.xml @@ -14,7 +14,6 @@ - - diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/ftp-message-history-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/ftp-message-history-context.xml index 526158d31a..06efe95287 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/ftp-message-history-context.xml +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/ftp-message-history-context.xml @@ -18,7 +18,6 @@ - lsEntries = this.channel.ls(path); @@ -90,28 +93,25 @@ class SftpSession implements Session { return new LsEntry[0]; } - public InputStream get(String source) { + public void copy(String source, OutputStream os) throws IOException{ Assert.state(this.channel != null, "session is not connected"); + try { - return this.channel.get(source); + InputStream is = this.channel.get(source); + FileCopyUtils.copy(is, os); } catch (SftpException e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to retrieve file", e); - } - return null; + throw new IOException("failed to copy file", e); } } - public void put(InputStream inputStream, String destination) { + public void copy(InputStream inputStream, String destination) throws IOException{ Assert.state(this.channel != null, "session is not connected"); try { this.channel.put(inputStream, destination); } catch (SftpException e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to copy file", e); - } + throw new IOException("failed to copy file", e); } }