From 5f36b01024790c8cf9b31a47aad97cc40b40d476 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 6 Dec 2010 12:25:07 -0500 Subject: [PATCH] INT-1664, INT-1665 added check for connection validity to recognize stale FTP connections, added 'throws IOException' to other relevant methods of Sessioin strategy, added debug statements to monitor session and session pool state, polished code --- .../remote/session/CachingSessionFactory.java | 42 +++++++++---------- .../file/remote/session/Session.java | 4 +- .../session/AbstractFtpSessionFactory.java | 1 - .../integration/ftp/session/FtpSession.java | 35 ++++++---------- .../integration/sftp/session/SftpSession.java | 13 ++---- 5 files changed, 38 insertions(+), 57 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 9e02b34779..206a464af5 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 @@ -21,8 +21,9 @@ import java.io.InputStream; import java.io.OutputStream; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.locks.ReentrantLock; -import java.util.logging.Logger; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; @@ -38,7 +39,7 @@ import org.springframework.beans.factory.DisposableBean; */ public class CachingSessionFactory implements SessionFactory, DisposableBean { - private static Logger logger = Logger.getLogger(CachingSessionFactory.class.getName()); + private static Log logger = LogFactory.getLog(CachingSessionFactory.class); public static final int DEFAULT_POOL_SIZE = 10; @@ -49,9 +50,6 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { private final int maxPoolSize; - private final ReentrantLock lock = new ReentrantLock(); - - public CachingSessionFactory(SessionFactory sessionFactory) { this(sessionFactory, DEFAULT_POOL_SIZE); } @@ -63,22 +61,20 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { } public Session getSession() { - this.lock.lock(); - try { - Session session = this.queue.poll(); + Session session = this.queue.poll(); + + if (session == null || (session != null && !session.isOpen())) { if (session != null && !session.isOpen()){ - this.queue.remove(session); - session = null; + logger.debug("Located session in the pool but it is stale, will create new one."); } - if (null == session) { - session = sessionFactory.getSession(); - } - - return (session != null) ? new CachedSession(session) : null; - } - finally { - this.lock.unlock(); + session = sessionFactory.getSession(); + logger.debug("Created new session"); } + else { + logger.debug("Using session from the pool"); + } + + return new CachedSession(session); } public void destroy() { @@ -97,7 +93,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { } catch (Throwable e) { // log and ignore - logger.warning("Exception was thrown while destroying Session. " + e); + logger.warn("Exception was thrown while destroying Session. ", e); } } @@ -112,18 +108,20 @@ public class CachingSessionFactory implements SessionFactory, DisposableBean { public void close() { if (queue.size() < maxPoolSize) { + logger.debug("Releasing target session back to the pool"); queue.add(targetSession); } else { + logger.debug("Disconnecting target session"); targetSession.close(); } } - public boolean remove(String path) { + public boolean remove(String path) throws IOException{ return this.targetSession.remove(path); } - public F[] list(String path) { + public F[] list(String path) throws IOException{ return this.targetSession.list(path); } 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 c0971500c7..2cd7c38438 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 @@ -31,9 +31,9 @@ import java.io.OutputStream; */ public interface Session { - boolean remove(String path); + boolean remove(String path) throws IOException; - F[] list(String path); + F[] list(String path) throws IOException; void copy(String source, OutputStream outputStream) throws IOException; 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 7ce9a8173f..a700532ab3 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 @@ -162,7 +162,6 @@ public abstract class AbstractFtpSessionFactory implements 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 ea3ba8c3da..4046483db2 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 @@ -48,35 +48,19 @@ class FtpSession implements Session { } - public boolean remove(String path) { + public boolean remove(String path) throws IOException { Assert.hasText(path, "path must not be null"); - boolean completed = false; - try { - completed = this.client.deleteFile(path); - if (!completed){ - throw new IOException("Failed to delete '" + path + "'. Server replied with: " + client.getReplyString()); - } - } - catch (IOException e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to delete file", e); - } + boolean completed = this.client.deleteFile(path); + if (!completed){ + throw new IOException("Failed to delete '" + path + "'. Server replied with: " + client.getReplyString()); } return completed; } @SuppressWarnings({"unchecked"}) - public FTPFile[] list(String path) { + public FTPFile[] list(String path) throws IOException { Assert.hasText(path, "path must not be null"); - try { - return this.client.listFiles(path); - } - catch (IOException e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to list files", e); - } - return new FTPFile[0]; - } + return this.client.listFiles(path); } public void copy(String path, OutputStream fos) throws IOException{ @@ -111,6 +95,11 @@ class FtpSession implements Session { } public boolean isOpen() { - return this.client.isConnected(); + try { + client.noop(); + } catch (Exception e) { + return false; + } + return true; } } 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 ccbc4a17da..50102593d7 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 @@ -57,22 +57,19 @@ class SftpSession implements Session { } - public boolean remove(String path) { + public boolean remove(String path) throws IOException{ Assert.state(this.channel != null, "session is not connected"); try { this.channel.rm(path); return true; } catch (SftpException e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to remove file", e); - } - return false; + throw new IOException("Failed to remove file", e); } } @SuppressWarnings("unchecked") - public LsEntry[] list(String path) { + public LsEntry[] list(String path) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { Vector lsEntries = this.channel.ls(path); @@ -87,9 +84,7 @@ class SftpSession implements Session { } } catch (SftpException e) { - if (logger.isWarnEnabled()) { - logger.warn("failed to list files", e); - } + throw new IOException("Failed to list files", e); } return new LsEntry[0]; }