From 583b2771cd9f0f28e98b27cba6a7df1fea4f7fb5 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 6 Dec 2010 09:06:29 -0500 Subject: [PATCH] INT-1664, INT-1665 added isOpen() method to the Session strategy, added check for stale connection to the CachingSessionFactory which will ensure that it only returns valid (non-stale) Session --- .../remote/session/CachingSessionFactory.java | 15 +++-- .../file/remote/session/Session.java | 2 + .../session/AbstractFtpSessionFactory.java | 57 +++++++------------ .../integration/ftp/session/FtpSession.java | 4 ++ .../integration/sftp/session/SftpSession.java | 5 ++ 5 files changed, 43 insertions(+), 40 deletions(-) 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 3a1fbbffb6..9e02b34779 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 @@ -25,7 +25,6 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Logger; import org.springframework.beans.factory.DisposableBean; -import org.springframework.util.Assert; /** * A {@link SessionFactory} implementation that caches Sessions for reuse without @@ -63,21 +62,23 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { this.queue = new ArrayBlockingQueue(this.maxPoolSize, true); } - public Session getSession() { - Assert.notNull(this.queue, "SftpSession is unavailable since the pool component is not started"); this.lock.lock(); try { Session session = this.queue.poll(); + if (session != null && !session.isOpen()){ + this.queue.remove(session); + session = null; + } if (null == session) { session = sessionFactory.getSession(); } + return (session != null) ? new CachedSession(session) : null; } finally { this.lock.unlock(); } - } public void destroy() { @@ -96,7 +97,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { } catch (Throwable e) { // log and ignore - logger.warning("Exception was thrown while destroying SftpSession. " + e); + logger.warning("Exception was thrown while destroying Session. " + e); } } @@ -133,6 +134,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { public void copy(InputStream inputStream, String destination) throws IOException{ this.targetSession.copy(inputStream, destination); } + + public boolean isOpen() { + return this.targetSession.isOpen(); + } } } 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 2e7dd839e1..c0971500c7 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 @@ -40,4 +40,6 @@ public interface Session { void copy(InputStream inputStream, String destination) throws IOException; void close(); + + boolean isOpen(); } 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 a7ed22e85e..7ce9a8173f 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 @@ -18,8 +18,6 @@ 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; @@ -27,7 +25,6 @@ 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; @@ -139,7 +136,6 @@ public abstract class AbstractFtpSessionFactory implements } } - @SuppressWarnings("unchecked") private T createClient() throws SocketException, IOException { final T client = this.createClientInstance(); Assert.notNull(client, "client must not be null"); @@ -148,37 +144,28 @@ public abstract class AbstractFtpSessionFactory implements 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(); + // Connect + client.connect(host); + + 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 + "]"); + + // Login + if (!client.login(username, password)) { + throw new IllegalStateException("Login failed. The respponse from the server is: " + + client.getReplyString()); + } + + this.postProcessClientAfterConnect(client); + + this.updateClientMode(client); + + client.setFileType(fileType); + client.setBufferSize(bufferSize); + return client; } /** diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index a3ed2dc339..ea3ba8c3da 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java @@ -109,4 +109,8 @@ class FtpSession implements Session { } } } + + public boolean isOpen() { + return this.client.isConnected(); + } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index acabb0c067..ccbc4a17da 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -39,6 +39,7 @@ import com.jcraft.jsch.SftpException; * @author Josh Long * @author Mario Gray * @author Mark Fisher + * @author Oleg Zhurakousky * @since 2.0 */ class SftpSession implements Session { @@ -136,4 +137,8 @@ class SftpSession implements Session { } } + public boolean isOpen() { + return this.channel.isConnected() && this.jschSession.isConnected(); + } + }