GH-9272: Close ClientSession from SftpSession

Fixes: #9272

* close ClientSession when closing SftpSession
* fix whitespace issues
* stop the SshClient on bean destruction
* use convenient assertions

(cherry picked from commit a3fb68a)

Author: darrylsmithUGA

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -49,6 +49,7 @@
 import org.apache.sshd.sftp.client.impl.AbstractSftpClient;
 import org.apache.sshd.sftp.client.impl.DefaultSftpClient;
 
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.core.io.Resource;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.file.remote.session.SessionFactory;
@@ -74,10 +75,11 @@
  * @author Auke Zaaiman
  * @author Christian Tzolov
  * @author Adama Sorho
+ * @author Darryl Smith
  *
  * @since 2.0
  */
-public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirEntry>, SharedSessionCapable {
+public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirEntry>, SharedSessionCapable, DisposableBean {
 
 	private final Lock lock = new ReentrantLock();
 
@@ -421,6 +423,13 @@ protected SftpClient createSftpClient(
 		return new ConcurrentSftpClient(clientSession, initialVersionSelector, errorDataHandler);
 	}
 
+	@Override
+	public void destroy() throws Exception {
+		if (this.isInnerClient && this.sshClient != null && this.sshClient.isStarted()) {
+			this.sshClient.stop();
+		}
+	}
+
 	/**
 	 * The {@link DefaultSftpClient} extension to lock the {@link #send(int, Buffer)}
 	 * for concurrent interaction.

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

@@ -25,6 +25,7 @@
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import org.apache.sshd.client.session.ClientSession;
 import org.apache.sshd.common.util.net.SshdSocketAddress;
 import org.apache.sshd.sftp.SftpModuleProperties;
 import org.apache.sshd.sftp.client.SftpClient;
@@ -47,6 +48,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Christian Tzolov
+ * @author Darryl Smith
  * @since 2.0
  */
 public class SftpSession implements Session<SftpClient.DirEntry> {
@@ -156,6 +158,16 @@ public void close() {
 		catch (IOException ex) {
 			throw new UncheckedIOException("failed to close an SFTP client", ex);
 		}
+
+		try {
+			ClientSession session = this.sftpClient.getSession();
+			if (session != null && session.isOpen()) {
+				session.close();
+			}
+		}
+		catch (IOException ex) {
+			throw new UncheckedIOException("failed to close an SFTP client (session)", ex);
+		}
 	}
 
 	@Override

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/dsl/SftpTests.java

@@ -31,6 +31,7 @@
 import org.junit.jupiter.api.condition.OS;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
@@ -43,6 +44,7 @@
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.sftp.SftpTestSupport;
 import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
@@ -59,6 +61,7 @@
  * @author Gary Russell
  * @author Joaquin Santana
  * @author Deepak Gunasekaran
+ * @author Darryl Smith
  *
  * @since 5.0
  */
@@ -69,11 +72,14 @@ public class SftpTests extends SftpTestSupport {
 	@Autowired
 	private IntegrationFlowContext flowContext;
 
+	@Autowired
+	private SessionFactory<SftpClient.DirEntry> sessionFactory;
+
 	@Test
 	public void testSftpInboundFlow() {
 		QueueChannel out = new QueueChannel();
 		IntegrationFlow flow = IntegrationFlow
-				.from(Sftp.inboundAdapter(sessionFactory())
+				.from(Sftp.inboundAdapter(sessionFactory)
 								.preserveTimestamp(true)
 								.remoteDirectory("/sftpSource")
 								.regexFilter(".*\\.txt$")
@@ -106,7 +112,7 @@ public void testSftpInboundFlow() {
 	public void testSftpInboundStreamFlow() throws Exception {
 		QueueChannel out = new QueueChannel();
 		StandardIntegrationFlow flow = IntegrationFlow.from(
-						Sftp.inboundStreamingAdapter(new SftpRemoteFileTemplate(sessionFactory()))
+						Sftp.inboundStreamingAdapter(new SftpRemoteFileTemplate(sessionFactory))
 								.remoteDirectory("sftpSource")
 								.regexFilter(".*\\.txt$"),
 						e -> e.id("sftpInboundAdapter").poller(Pollers.fixedDelay(100)))
@@ -133,7 +139,7 @@ public void testSftpInboundStreamFlow() throws Exception {
 
 	@Test
 	public void testSftpOutboundFlow() {
-		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sessionFactory(), FileExistsMode.FAIL)
+		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sessionFactory, FileExistsMode.FAIL)
 				.useTemporaryFileName(false)
 				.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
 				.remoteDirectory("sftpTarget"));
@@ -143,7 +149,7 @@ public void testSftpOutboundFlow() {
 				.setHeader(FileHeaders.FILENAME, fileName)
 				.build());
 
-		RemoteFileTemplate<SftpClient.DirEntry> template = new RemoteFileTemplate<>(sessionFactory());
+		RemoteFileTemplate<SftpClient.DirEntry> template = new RemoteFileTemplate<>(sessionFactory);
 		SftpClient.DirEntry[] files =
 				template.execute(session -> session.list(getTargetRemoteDirectory().getName() + "/" + fileName));
 		assertThat(files.length).isEqualTo(1);
@@ -154,7 +160,7 @@ public void testSftpOutboundFlow() {
 
 	@Test
 	public void testSftpOutboundFlowSftpTemplate() {
-		SftpRemoteFileTemplate sftpTemplate = new SftpRemoteFileTemplate(sessionFactory());
+		SftpRemoteFileTemplate sftpTemplate = new SftpRemoteFileTemplate(sessionFactory);
 		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sftpTemplate)
 				.useTemporaryFileName(false)
 				.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
@@ -175,7 +181,7 @@ public void testSftpOutboundFlowSftpTemplate() {
 
 	@Test
 	public void testSftpOutboundFlowSftpTemplateAndMode() {
-		SftpRemoteFileTemplate sftpTemplate = new SftpRemoteFileTemplate(sessionFactory());
+		SftpRemoteFileTemplate sftpTemplate = new SftpRemoteFileTemplate(sessionFactory);
 		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sftpTemplate, FileExistsMode.APPEND)
 				.useTemporaryFileName(false)
 				.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
@@ -201,7 +207,7 @@ public void testSftpOutboundFlowSftpTemplateAndMode() {
 	@Test
 	@DisabledOnOs(OS.WINDOWS)
 	public void testSftpOutboundFlowWithChmod() {
-		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sessionFactory(), FileExistsMode.FAIL)
+		IntegrationFlow flow = f -> f.handle(Sftp.outboundAdapter(sessionFactory, FileExistsMode.FAIL)
 				.useTemporaryFileName(false)
 				.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
 				.chmod(0644)
@@ -212,7 +218,7 @@ public void testSftpOutboundFlowWithChmod() {
 				.setHeader(FileHeaders.FILENAME, fileName)
 				.build());
 
-		RemoteFileTemplate<SftpClient.DirEntry> template = new RemoteFileTemplate<>(sessionFactory());
+		RemoteFileTemplate<SftpClient.DirEntry> template = new RemoteFileTemplate<>(sessionFactory);
 		SftpClient.DirEntry[] files =
 				template.execute(session -> session.list(getTargetRemoteDirectory().getName() + "/" + fileName));
 		assertThat(files.length).isEqualTo(1);
@@ -231,7 +237,7 @@ public void testSftpOutboundFlowWithChmod() {
 	public void testSftpMgetFlow() {
 		QueueChannel out = new QueueChannel();
 		IntegrationFlow flow = f -> f
-				.handle(Sftp.outboundGateway(sessionFactory(), AbstractRemoteFileOutboundGateway.Command.MGET,
+				.handle(Sftp.outboundGateway(sessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
 								"payload")
 						.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
 						.regexFileNameFilter("(subSftpSource|.*1.txt)")
@@ -260,7 +266,7 @@ public void testSftpMgetFlow() {
 	public void testSftpSessionCallback() {
 		QueueChannel out = new QueueChannel();
 		IntegrationFlow flow = f -> f
-				.<String>handle((p, h) -> new SftpRemoteFileTemplate(sessionFactory()).execute(s -> s.list(p)))
+				.<String>handle((p, h) -> new SftpRemoteFileTemplate(sessionFactory).execute(s -> s.list(p)))
 				.channel(out);
 		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
 		registration.getInputChannel().send(new GenericMessage<>("sftpSource"));
@@ -279,7 +285,7 @@ public void testSftpSessionCallback() {
 	public void testSftpMv() {
 		QueueChannel out = new QueueChannel();
 		IntegrationFlow flow = f -> f
-				.handle(Sftp.outboundGateway(sessionFactory(), AbstractRemoteFileOutboundGateway.Command.MV, "payload")
+				.handle(Sftp.outboundGateway(sessionFactory, AbstractRemoteFileOutboundGateway.Command.MV, "payload")
 						.renameExpression("payload.concat('.done')")
 						.remoteDirectoryExpression("'sftpSource'"))
 				.channel(out);
@@ -303,6 +309,11 @@ public void testSftpMv() {
 	@EnableIntegration
 	public static class ContextConfiguration {
 
+		@Bean
+		public SessionFactory<SftpClient.DirEntry> ftpsessionFactory() {
+			return SftpTests.sessionFactory();
+		}
+
 	}
 
 }

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

@@ -61,6 +61,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Joaquin Santana
+ * @author Darryl Smith
  *
  * @since 2.0
  */
@@ -160,6 +161,8 @@ public void testCopyFileToLocalDir() throws Exception {
 		ms.stop();
 		verify(synchronizer).close();
 		verify(store).close();
+
+		ftpSessionFactory.destroy();
 	}
 
 	public static class TestSftpSessionFactory extends DefaultSftpSessionFactory {

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpMessageSourceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 the original author or authors.
+ * Copyright 2018-2024 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.
@@ -16,12 +16,15 @@
 
 package org.springframework.integration.sftp.inbound;
 
+import org.apache.sshd.sftp.client.SftpClient;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.sftp.SftpTestSupport;
 import org.springframework.messaging.Message;
 import org.springframework.test.annotation.DirtiesContext;
@@ -32,6 +35,7 @@
 /**
  * @author Gary Russell
  * @author Artem bilan
+ * @author Darryl Smith
  *
  * @since 5.0.7
  *
@@ -43,6 +47,9 @@ public class SftpMessageSourceTests extends SftpTestSupport {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	private SessionFactory<SftpClient.DirEntry> sessionFactory;
+
 	@Test
 	public void testMaxFetch() {
 		SftpInboundFileSynchronizingMessageSource messageSource = buildSource();
@@ -53,7 +60,7 @@ public void testMaxFetch() {
 	}
 
 	private SftpInboundFileSynchronizingMessageSource buildSource() {
-		SftpInboundFileSynchronizer sync = new SftpInboundFileSynchronizer(sessionFactory());
+		SftpInboundFileSynchronizer sync = new SftpInboundFileSynchronizer(sessionFactory);
 		sync.setRemoteDirectory("/sftpSource/");
 		sync.setBeanFactory(this.context);
 		SftpInboundFileSynchronizingMessageSource messageSource = new SftpInboundFileSynchronizingMessageSource(sync);
@@ -68,6 +75,11 @@ private SftpInboundFileSynchronizingMessageSource buildSource() {
 	@Configuration
 	public static class Config {
 
+		@Bean
+		public SessionFactory<SftpClient.DirEntry> ftpSessionFactory() {
+			return SftpMessageSourceTests.sessionFactory();
+		}
+
 	}
 
 }

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java

@@ -76,6 +76,7 @@
  * @author Gary Russell
  * @author Gunnar Hillert
  * @author Artem Bilan
+ * @author Darryl Smith
  */
 public class SftpOutboundTests {
 
@@ -84,7 +85,7 @@ public void testHandleFileMessage() throws Exception {
 		File targetDir = new File("remote-target-dir");
 		assertThat(targetDir.exists()).as("target directory does not exist: " + targetDir.getName()).isTrue();
 
-		SessionFactory<SftpClient.DirEntry> sessionFactory = new TestSftpSessionFactory();
+		TestSftpSessionFactory sessionFactory = new TestSftpSessionFactory();
 		FileTransferringMessageHandler<SftpClient.DirEntry> handler =
 				new FileTransferringMessageHandler<>(sessionFactory);
 		handler.setRemoteDirectoryExpression(new LiteralExpression(targetDir.getName()));
@@ -103,6 +104,8 @@ public void testHandleFileMessage() throws Exception {
 
 		handler.handleMessage(new GenericMessage<>(srcFile));
 		assertThat(destFile.exists()).as("destination file was not created").isTrue();
+
+		sessionFactory.destroy();
 	}
 
 	@Test
@@ -111,7 +114,7 @@ public void testHandleStringMessage() throws Exception {
 		if (file.exists()) {
 			file.delete();
 		}
-		SessionFactory<SftpClient.DirEntry> sessionFactory = new TestSftpSessionFactory();
+		TestSftpSessionFactory sessionFactory = new TestSftpSessionFactory();
 		FileTransferringMessageHandler<SftpClient.DirEntry> handler =
 				new FileTransferringMessageHandler<>(sessionFactory);
 		DefaultFileNameGenerator fGenerator = new DefaultFileNameGenerator();
@@ -127,6 +130,8 @@ public void testHandleStringMessage() throws Exception {
 		byte[] inFile = FileCopyUtils.copyToByteArray(file);
 		assertThat(new String(inFile)).isEqualTo("String data");
 		file.delete();
+
+		sessionFactory.destroy();
 	}
 
 	@Test
@@ -135,7 +140,7 @@ public void testHandleBytesMessage() throws Exception {
 		if (file.exists()) {
 			file.delete();
 		}
-		SessionFactory<SftpClient.DirEntry> sessionFactory = new TestSftpSessionFactory();
+		TestSftpSessionFactory sessionFactory = new TestSftpSessionFactory();
 		FileTransferringMessageHandler<SftpClient.DirEntry> handler =
 				new FileTransferringMessageHandler<>(sessionFactory);
 		DefaultFileNameGenerator fGenerator = new DefaultFileNameGenerator();
@@ -151,6 +156,8 @@ public void testHandleBytesMessage() throws Exception {
 		byte[] inFile = FileCopyUtils.copyToByteArray(file);
 		assertThat(new String(inFile)).isEqualTo("byte[] data");
 		file.delete();
+
+		sessionFactory.destroy();
 	}
 
 	@Test //INT-2275
@@ -225,7 +232,7 @@ public void testMkDir() throws Exception {
 
 	@ParameterizedTest
 	@ValueSource(booleans = {true, false})
-	public void testSharedSession(boolean sharedSession) throws IOException {
+	public void testSharedSession(boolean sharedSession) throws Exception {
 		try (SshServer server = SshServer.setUpDefaultServer()) {
 			server.setPasswordAuthenticator((arg0, arg1, arg2) -> true);
 			server.setPort(0);
@@ -253,6 +260,8 @@ public void testSharedSession(boolean sharedSession) throws IOException {
 				assertThat(TestUtils.getPropertyValue(s2, "sftpClient"))
 						.isNotSameAs(TestUtils.getPropertyValue(s1, "sftpClient"));
 			}
+
+			f.destroy();
 		}
 	}