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

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> F[] list(String path) {
+		public <F> F[] list(String path) throws IOException{
 			return this.targetSession.<F>list(path);
 		}
 

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> F[] list(String path);
+	<F> F[] list(String path) throws IOException;
 	
 	void copy(String source, OutputStream outputStream) throws IOException;
 

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/AbstractFtpSessionFactory.java

@@ -162,7 +162,6 @@ public abstract class AbstractFtpSessionFactory<T extends FTPClient> implements
 		this.postProcessClientAfterConnect(client);
 		
 		this.updateClientMode(client);
-		
 		client.setFileType(fileType);
 		client.setBufferSize(bufferSize);
 		return client;

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

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