GH-9380: Add DefaultSftpSessionFactory.setSshClientConfigurer()

Fixes: #9380

Expose a `Consumer<SshClient> sshClientConfigurer` option for the `DefaultSftpSessionFactory`
to further customize an internal `SshClient` instance.

Author: artembilan

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

@@ -24,6 +24,7 @@
 import java.util.Collection;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 
 import org.apache.sshd.client.SshClient;
 import org.apache.sshd.client.auth.keyboard.UserInteraction;
@@ -79,7 +80,8 @@
  *
  * @since 2.0
  */
-public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirEntry>, SharedSessionCapable, DisposableBean {
+public class DefaultSftpSessionFactory
+		implements SessionFactory<SftpClient.DirEntry>, SharedSessionCapable, DisposableBean {
 
 	private final Lock lock = new ReentrantLock();
 
@@ -119,6 +121,9 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 
 	private volatile SftpClient sharedSftpClient;
 
+	private Consumer<SshClient> sshClientConfigurer = (sshClient) -> {
+	};
+
 	public DefaultSftpSessionFactory() {
 		this(false);
 	}
@@ -283,6 +288,20 @@ public void setSftpVersionSelector(SftpVersionSelector sftpVersionSelector) {
 		this.sftpVersionSelector = sftpVersionSelector;
 	}
 
+	/**
+	 * Set a {@link Consumer} as a callback to further customize an internal {@link SshClient} instance.
+	 * For example, to set custom values for its properties using {@link PropertyResolverUtils#updateProperty} API.
+	 * @param sshClientConfigurer the {@link Consumer} to configure an internal {@link SshClient} instance.
+	 * @since 6.4
+	 * @see SshClient
+	 * @see PropertyResolverUtils#updateProperty
+	 */
+	public void setSshClientConfigurer(Consumer<SshClient> sshClientConfigurer) {
+		Assert.state(this.isInnerClient, "Cannot mutate externally provided SshClient");
+		Assert.notNull(sshClientConfigurer, "'sshClientConfigurer' must noy be null");
+		this.sshClientConfigurer = sshClientConfigurer;
+	}
+
 	@Override
 	public SftpSession getSession() {
 		SftpSession sftpSession;
@@ -392,6 +411,7 @@ public InputStream openInputStream() throws IOException {
 				}
 			}
 			this.sshClient.setUserInteraction(this.userInteraction);
+			this.sshClientConfigurer.accept(this.sshClient);
 		}
 	}
 
@@ -424,7 +444,7 @@ protected SftpClient createSftpClient(
 	}
 
 	@Override
-	public void destroy() throws Exception {
+	public void destroy() {
 		if (this.isInnerClient && this.sshClient != null && this.sshClient.isStarted()) {
 			this.sshClient.stop();
 		}

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java

@@ -31,7 +31,9 @@
 import org.apache.sshd.client.channel.ClientChannel;
 import org.apache.sshd.client.keyverifier.AcceptAllServerKeyVerifier;
 import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.PropertyResolverUtils;
 import org.apache.sshd.common.SshException;
+import org.apache.sshd.core.CoreModuleProperties;
 import org.apache.sshd.server.SshServer;
 import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
 import org.apache.sshd.sftp.client.SftpClient;
@@ -206,7 +208,7 @@ void concurrentSessionListDoesntCauseFailure() throws Exception {
 	}
 
 	@Test
-	void customTimeoutIsApplied() throws Exception {
+	void customPropertiesAreApplied() throws Exception {
 		try (SshServer server = SshServer.setUpDefaultServer()) {
 			server.setPasswordAuthenticator((arg0, arg1, arg2) -> true);
 			server.setPort(0);
@@ -221,10 +223,15 @@ void customTimeoutIsApplied() throws Exception {
 			sftpSessionFactory.setPassword("pass");
 			sftpSessionFactory.setAllowUnknownKeys(true);
 			sftpSessionFactory.setTimeout(15_000);
+			sftpSessionFactory.setSshClientConfigurer((sshClient) -> {
+				sshClient.setNioWorkers(27);
+				PropertyResolverUtils.updateProperty(sshClient, CoreModuleProperties.MAX_PACKET_SIZE.getName(), 48 * 1024);
+			});
 
 			ClientChannel clientChannel = sftpSessionFactory.getSession().getClientInstance().getClientChannel();
 
 			assertThat(AbstractSftpClient.SFTP_CLIENT_CMD_TIMEOUT.getRequired(clientChannel)).hasSeconds(15);
+			assertThat(CoreModuleProperties.MAX_PACKET_SIZE.getRequired(clientChannel)).isEqualTo(48 * 1024);
 
 			sftpSessionFactory.destroy();
 		}
@@ -260,4 +267,5 @@ void clientSessionIsClosedOnSessionClose() throws Exception {
 			sftpSessionFactory.destroy();
 		}
 	}
+
 }

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

@@ -107,3 +107,13 @@ If `false`, a pre-populated `knownHosts` file is required.
 
 `userInteraction`::A custom `org.apache.sshd.client.auth.keyboard.UserInteraction` to be used during authentication.
 
+Starting with version 6.4, the `DefaultSftpSessionFactory` expose a `Consumer<SshClient>` configurer property to further customize an internal `SshClient`.
+For example, this is how to change a default number of NIO workers and packet size for the client:
+
+[source, java]
+----
+sftpSessionFactory.setSshClientConfigurer((sshClient) -> {
+    sshClient.setNioWorkers(27);
+    PropertyResolverUtils.updateProperty(sshClient, CoreModuleProperties.MAX_PACKET_SIZE.getName(), 48 * 1024);
+});
+----

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

@@ -59,4 +59,10 @@ See xref:redis.adoc[Redis Support] for more information.
 === Groovy Changes
 
 The `ControlBusFactoryBean` (and respective `<int-groovy:control-bus>` XML tag) has been deprecated (for removal) in favor of new introduced `ControlBusFactoryBean` based on a new model implemented in the `ControlBusCommandRegistry`.
-See xref:control-bus.adoc[Control Bus] for more information.
+See xref:control-bus.adoc[Control Bus] for more information.
+
+[[x6.4-sftp-changes]]
+=== SFTP Support Changes
+
+The `DefaultSftpSessionFactory` now exposes a `Consumer<SshClient>` configurer property to further customize an internal `SshClient`.
+See xref:sftp/session-factory.adoc[SFTP Session Factory] for more information.