GH-8713: Add support for custom SftpClient

Fixes https://github.com/spring-projects/spring-integration/issues/8713

* Introduce `DefaultSftpSessionFactory.createSftpClient()` factory method
which can be overridden for any custom `SftpClient` use-case
* Add changes to the docs
* Some code clean up

**Cherry-pick to `6.1.x`**

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

@@ -53,6 +53,10 @@ import org.springframework.util.Assert;
 
 /**
  * Factory for creating {@link SftpSession} instances.
+ * <p>
+ * The {@link #createSftpClient(ClientSession, SftpVersionSelector, SftpErrorDataHandler)}
+ * can be overridden to provide a custom {@link SftpClient}.
+ * The {@link ConcurrentSftpClient} is used by default.
  *
  * @author Josh Long
  * @author Mario Gray
@@ -65,6 +69,7 @@ import org.springframework.util.Assert;
  * @author Krzysztof Debski
  * @author Auke Zaaiman
  * @author Christian Tzolov
+ * @author Adama Sorho
  *
  * @since 2.0
  */
@@ -282,9 +287,7 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 		try {
 			boolean freshSftpClient = false;
 			if (sftpClient == null || !sftpClient.isOpen()) {
-				sftpClient =
-						new ConcurrentSftpClient(initClientSession(), this.sftpVersionSelector,
-								SftpErrorDataHandler.EMPTY);
+				sftpClient = createSftpClient(initClientSession(), this.sftpVersionSelector, SftpErrorDataHandler.EMPTY);
 				freshSftpClient = true;
 			}
 			sftpSession = new SftpSession(sftpClient);
@@ -293,8 +296,8 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 				this.sharedSftpClient = sftpClient;
 			}
 		}
-		catch (Exception e) {
-			throw new IllegalStateException("failed to create SFTP Session", e);
+		catch (Exception ex) {
+			throw new IllegalStateException("failed to create SFTP Session", ex);
 		}
 		finally {
 			if (this.sharedSessionLock != null) {
@@ -397,15 +400,32 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 		this.sharedSftpClient = null;
 	}
 
+	/**
+	 * Can be overridden to provide a custom {@link SftpClient} to {@link #getSession()}.
+	 * @param clientSession the {@link ClientSession}
+	 * @param initialVersionSelector the initial {@link SftpVersionSelector}
+	 * @param errorDataHandler the {@link SftpErrorDataHandler} to handle incoming data
+	 *                         through the error stream.
+	 * @return {@link SftpClient}
+	 * @throws IOException if failed to initialize
+	 * @since 6.1.3
+	 */
+	protected SftpClient createSftpClient(
+			ClientSession clientSession, SftpVersionSelector initialVersionSelector,
+			SftpErrorDataHandler errorDataHandler) throws IOException {
+
+		return new ConcurrentSftpClient(clientSession, initialVersionSelector, errorDataHandler);
+	}
+
 	/**
 	 * The {@link DefaultSftpClient} extension to lock the {@link #send(int, Buffer)}
 	 * for concurrent interaction.
 	 */
-	private static class ConcurrentSftpClient extends DefaultSftpClient {
+	protected static class ConcurrentSftpClient extends DefaultSftpClient {
 
 		private final Lock sendLock = new ReentrantLock();
 
-		ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
+		protected ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
 				SftpErrorDataHandler errorDataHandler) throws IOException {
 
 			super(clientSession, initialVersionSelector, errorDataHandler);

src/reference/antora/modules/ROOT/pages/sftp/session-factory.adoc

@@ -38,13 +38,26 @@ When using this feature, you must wrap the session factory in a caching session
 
 If the cache is reset, the session is disconnected only when the last channel is closed.
 
-The connection is refreshed if it is found to be disconnected when a new operation obtains a session.
+The connection is refreshed if it to be disconnected when a new operation obtains a session.
 =====
 
 Now all you need to do is inject this SFTP session factory into your adapters.
 
 NOTE: A more practical way to provide values for the SFTP session factory is to use Spring's https://docs.spring.io/spring/docs/current/spring-framework-reference/core.html#beans-factory-placeholderconfigurer[property placeholder support].
 
+Starting with version 6.1.3, the `DefaultSftpSessionFactory` introduces a `createSftpClient(...)` to support a custom `SftpClient`.
+See a sample below of how to override `createSftpChannelSubsystem()` method in your custom `SftpClient` to add, for example, some custom `RequestHandler` for SFTP sub-system requests and replies:
+
+[source, java]
+----
+@Override
+protected ChannelSubsystem createSftpChannelSubsystem(ClientSession clientSession) {
+    ChannelSubsystem sftpChannelSubsystem = super.createSftpChannelSubsystem(clientSession);
+    sftpChannelSubsystem.addRequestHandler((channel, request, wantReply, buffer) -> ...);
+    return sftpChannelSubsystem;
+}
+----
+
 [[sftp-session-factory-properties]]
 == Configuration Properties
 

src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -72,3 +72,9 @@ See xref:mongodb.adoc#mongodb-message-store[MongoDB Message Store] for an exampl
 
 `FtpLastModifiedFileListFilter`, `SftpLastModifiedFileListFilter` and `SmbLastModifiedFileListFilter` have been introduced to allow files filtering based on a last-modified strategy respectively for `FTP`, `SFTP` and `SMB`.
 See xref:ftp/inbound.adoc#ftp-inbound[FTP Inbound Channel Adapter], xref:sftp/inbound.adoc#sftp-inbound[SFTP Inbound Channel Adapter], and xref:smb.adoc#smb-inbound[SMB Inbound Channel Adapter] for more information.
+
+[[x6.2-sftp-changes]]
+=== SFTP Support Changes
+
+A new `DefaultSftpSessionFactory.createSftpClient(...)` method has been introduced to support a custom `SftpClient` when overridden.
+See xref:sftp/session-factory.adoc#sftp-session-factory[SFTP Session Factory] for more information.