GH-2605: (S)FTP test cached sessions

Resolves #2605

* Suppress unused field warning.

* Fix typos.

* Use lstat() - don't follow symbolic lincs

Author: garyrussell

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,6 +48,8 @@ public class CachingSessionFactory<F> implements SessionFactory<F>, DisposableBe
 
 	private final boolean isSharedSessionCapable;
 
+	private boolean testSession;
+
 	private volatile long sharedSessionEpoch;
 
 	/**
@@ -83,7 +85,7 @@ public Session<F> createForPool() {
 
 			@Override
 			public boolean isStale(Session<F> session) {
-				return !session.isOpen();
+				return CachingSessionFactory.this.testSession ? !session.test() : !session.isOpen();
 			}
 
 			@Override
@@ -115,6 +117,15 @@ public void setPoolSize(int poolSize) {
 		this.pool.setPoolSize(poolSize);
 	}
 
+	/**
+	 * Set to true to test the session when checking one out from the cache.
+	 * @param testSession true to test.
+	 * @since 5.1
+	 */
+	public void setTestSession(boolean testSession) {
+		this.testSession = testSession;
+	}
+
 	/**
 	 * Get a session from the pool (or block if none available).
 	 */

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -107,4 +107,14 @@ public interface Session<F> extends Closeable {
 	 */
 	Object getClientInstance();
 
+	/**
+	 * Test the session is still alive, e.g. when checking out from a pool.
+	 * The default implementation simply delegates to {@link #isOpen()}.
+	 * @return true if the test is successful.
+	 * @since 5.1
+	 */
+	default boolean test() {
+		return this.isOpen();
+	}
+
 }

spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -52,6 +52,7 @@ public class CachingSessionFactoryTests {
 	public void testCacheAndReset() {
 		TestSessionFactory factory = new TestSessionFactory();
 		CachingSessionFactory<String> cache = new CachingSessionFactory<String>(factory);
+		cache.setTestSession(true);
 		Session<String> sess1 = cache.getSession();
 		assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id"));
 		Session<String> sess2 = cache.getSession();
@@ -61,6 +62,7 @@ public void testCacheAndReset() {
 		assertTrue(sess1.isOpen());
 		sess1 = cache.getSession();
 		assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id"));
+		assertTrue((TestUtils.getPropertyValue(sess1, "targetSession.testCalled", Boolean.class)));
 		sess1.close();
 		assertTrue(sess1.isOpen());
 		// reset the cache; should close idle (sess1); sess2 should closed later
@@ -122,6 +124,9 @@ private class TestSession implements Session<String> {
 
 		private volatile boolean open = true;
 
+		@SuppressWarnings("unused")
+		private boolean testCalled;
+
 		private TestSession(String id) {
 			this.id = id;
 		}
@@ -197,6 +202,12 @@ public Object getClientInstance() {
 			return null;
 		}
 
+		@Override
+		public boolean test() {
+			this.testCalled = true;
+			return true;
+		}
+
 	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -226,4 +226,20 @@ public FTPClient getClientInstance() {
 		return this.client;
 	}
 
+
+	@Override
+	public boolean test() {
+		return isOpen() && doTest();
+	}
+
+	private boolean doTest() {
+		try {
+			this.client.noop();
+			return true;
+		}
+		catch (IOException e) {
+			return false;
+		}
+	}
+
 }

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -278,4 +278,20 @@ public ChannelSftp getClientInstance() {
 		return this.channel;
 	}
 
+	@Override
+	public boolean test() {
+		return isOpen() && doTest();
+	}
+
+	private boolean doTest() {
+		try {
+			this.channel.lstat(this.channel.getHome());
+			return true;
+		}
+		catch (Exception e) {
+			return false;
+		}
+	}
+
+
 }

src/reference/asciidoc/ftp.adoc

@@ -85,10 +85,6 @@ The following example shows a complete configuration:
 ----
 ====
 
-Every time an adapter requests a session object from its `SessionFactory`, the session is returned from a session pool maintained by a caching wrapper around the factory.
-A session in the session pool might go stale (if it has been disconnected by the server due to inactivity), so the `SessionFactory` performs validation to make sure that it never returns a stale session to the adapter.
-If a stale session was encountered, it is removed from the pool, and a new one is created.
-
 NOTE: If you experience connectivity problems and would like to trace session creation as well as see which sessions are polled, you can enable session tracing by setting the logger to the `TRACE` level (for example, `log4j.category.org.springframework.integration.file=TRACE`).
 
 Now you need only inject these session factories into your adapters.
@@ -482,6 +478,7 @@ public class FtpJavaApplication {
         sf.setPort(port);
         sf.setUsername("foo");
         sf.setPassword("foo");
+        sf.setTestSession(true);
         return new CachingSessionFactory<FTPFile>(sf);
     }
 
@@ -884,6 +881,7 @@ public class FtpJavaApplication {
         sf.setPort(port);
         sf.setUsername("foo");
         sf.setPassword("foo");
+        sf.setTestSession(true);
         return new CachingSessionFactory<FTPFile>(sf);
     }
 
@@ -941,6 +939,7 @@ public class FtpJavaApplication {
         sf.setPort(port);
         sf.setUsername("foo");
         sf.setPassword("foo");
+        sf.setTestSession(true);
         return new CachingSessionFactory<FTPFile>(sf);
     }
 
@@ -1274,6 +1273,7 @@ public class FtpJavaApplication {
         sf.setPort(port);
         sf.setUsername("foo");
         sf.setPassword("foo");
+        sf.setTestSession(true);
         return new CachingSessionFactory<FTPFile>(sf);
     }
 
@@ -1313,6 +1313,7 @@ public class FtpJavaApplication {
         sf.setPort(port);
         sf.setUsername("foo");
         sf.setPassword("foo");
+        sf.setTestSession(true);
         return new CachingSessionFactory<FTPFile>(sf);
     }
 
@@ -1412,6 +1413,9 @@ Starting with Spring Integration 3.0, the `CachingConnectionFactory` provides a
 When invoked, all idle sessions are immediately closed and in-use sessions are closed when they are returned to the cache.
 New requests for sessions establish new sessions as necessary.
 
+Starting with version 5.1, the `CachingSessionFactory` has a new property `testSession`.
+When true, the session will be tested by sending a NOOP command to ensure it is still active; if not, it will be removed from the cache; a new session is created if no active sessions are in the cache.
+
 [[ftp-rft]]
 === Using `RemoteFileTemplate`
 

src/reference/asciidoc/sftp.adoc

@@ -253,6 +253,9 @@ When invoked, all idle sessions are immediately closed and in-use sessions are c
 When using `isSharedSession=true`, the channel is closed and the shared session is closed only when the last channel is closed.
 New requests for sessions establish new sessions as necessary.
 
+Starting with version 5.1, the `CachingSessionFactory` has a new property `testSession`.
+When true, the session will be tested by performing a `stat(getHome())` command to ensure it is still active; if not, it will be removed from the cache; a new session is created if no active sessions are in the cache.
+
 [[sftp-rft]]
 === Using `RemoteFileTemplate`
 
@@ -476,6 +479,7 @@ public class SftpJavaApplication {
         factory.setUser("foo");
         factory.setPassword("foo");
         factory.setAllowUnknownKeys(true);
+        factory.setTestSession(true);
         return new CachingSessionFactory<LsEntry>(factory);
     }
 
@@ -873,6 +877,7 @@ public class SftpJavaApplication {
         factory.setUser("foo");
         factory.setPassword("foo");
         factory.setAllowUnknownKeys(true);
+        factory.setTestSession(true);
         return new CachingSessionFactory<LsEntry>(factory);
     }
 
@@ -1257,6 +1262,7 @@ public class SftpJavaApplication {
         sf.setPort(port);
         sf.setUsername("foo");
         sf.setPassword("foo");
+        factory.setTestSession(true);
         return new CachingSessionFactory<LsEntry>(sf);
     }
 

src/reference/asciidoc/whats-new.adoc

@@ -162,6 +162,10 @@ See <<ftp-streaming>> and <<sftp-streaming>> for more information.
 In addition, the synchronizers for inbound channel adapters can now be provided with a `Comparator`.
 This is useful when using `maxFetchSize` to limit the files retrieved.
 
+The `CachingSessionFactory` has a new property `testSession` which, when true, causes the factory to perform a `test()` operation on the `Session` when checking out an existing session from the cache.
+
+See <<sftp-session-caching>> and <<ftp-session-caching>> for more information.
+
 [[x51.-tcp]]
 === TCP Support