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]; }