diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 217e74e889..175833c114 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -18,6 +18,7 @@ package org.springframework.integration.sftp.session; import java.io.IOException; import java.io.InputStream; +import java.io.UncheckedIOException; import java.security.GeneralSecurityException; import java.security.KeyPair; import java.time.Duration; @@ -316,7 +317,7 @@ public class DefaultSftpSessionFactory sftpClient = createSftpClient(initClientSession(), this.sftpVersionSelector, SftpErrorDataHandler.EMPTY); freshSftpClient = true; } - sftpSession = new SftpSession(sftpClient); + sftpSession = new SftpSession(sftpClient, this.isSharedSession); sftpSession.connect(); if (this.isSharedSession && freshSftpClient) { this.sharedSftpClient = sftpClient; @@ -449,6 +450,26 @@ public class DefaultSftpSessionFactory if (this.isInnerClient && this.sshClient != null && this.sshClient.isStarted()) { this.sshClient.stop(); } + + SftpClient sharedSftpClientToClose = this.sharedSftpClient; + if (sharedSftpClientToClose != null) { + try { + sharedSftpClientToClose.close(); + } + catch (IOException ex) { + throw new UncheckedIOException("failed to close an SFTP client", ex); + } + + try { + ClientSession session = sharedSftpClientToClose.getSession(); + if (session != null && session.isOpen()) { + session.close(); + } + } + catch (IOException ex) { + throw new UncheckedIOException("failed to close an SFTP client (session)", ex); + } + } } /** diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index 860cabea9e..9255fc4e25 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -55,9 +55,23 @@ public class SftpSession implements Session { private final SftpClient sftpClient; + private final boolean isSharedClient; + public SftpSession(SftpClient sftpClient) { + this(sftpClient, false); + } + + /** + * Construct an instance based on a {@link SftpClient} and its {@code shared} status. + * When {@code isSharedClient == true}, the {@link #close()} is void. + * @param sftpClient the {@link SftpClient} to use. + * @param isSharedClient whether the {@link SftpClient} is shared. + * @since 6.3.9 + */ + public SftpSession(SftpClient sftpClient, boolean isSharedClient) { Assert.notNull(sftpClient, "'sftpClient' must not be null"); this.sftpClient = sftpClient; + this.isSharedClient = isSharedClient; } @Override @@ -152,6 +166,10 @@ public class SftpSession implements Session { @Override public void close() { + if (this.isSharedClient) { + return; + } + try { this.sftpClient.close(); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java index 05d245df5b..3c58b877e0 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java @@ -24,6 +24,11 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; import org.apache.sshd.client.SshClient; @@ -37,6 +42,8 @@ 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; +import org.apache.sshd.sftp.client.SftpErrorDataHandler; +import org.apache.sshd.sftp.client.SftpVersionSelector; import org.apache.sshd.sftp.client.impl.AbstractSftpClient; import org.apache.sshd.sftp.server.SftpSubsystemFactory; import org.junit.jupiter.api.Test; @@ -268,4 +275,64 @@ public class SftpSessionFactoryTests { } } + @Test + void sharedSessionConcurrentAccess() throws Exception { + try (SshServer server = SshServer.setUpDefaultServer()) { + server.setPasswordAuthenticator((arg0, arg1, arg2) -> true); + server.setPort(0); + server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File("hostkey.ser").toPath())); + server.setSubsystemFactories(Collections.singletonList(new SftpSubsystemFactory())); + server.start(); + + AtomicInteger clientInstances = new AtomicInteger(); + + DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(true) { + + @Override + protected SftpClient createSftpClient(ClientSession clientSession, + SftpVersionSelector initialVersionSelector, SftpErrorDataHandler errorDataHandler) + throws IOException { + + clientInstances.incrementAndGet(); + return super.createSftpClient(clientSession, initialVersionSelector, errorDataHandler); + } + + }; + sftpSessionFactory.setHost("localhost"); + sftpSessionFactory.setPort(server.getPort()); + sftpSessionFactory.setUser("user"); + sftpSessionFactory.setPassword("pass"); + sftpSessionFactory.setAllowUnknownKeys(true); + + ExecutorService executorService = Executors.newFixedThreadPool(10); + + CountDownLatch executionLatch = new CountDownLatch(20); + List errors = Collections.synchronizedList(new ArrayList<>()); + + for (int i = 0; i < 20; i++) { + executorService.execute(() -> { + try (SftpSession session = sftpSessionFactory.getSession()) { + session.list("."); + } + catch (Exception e) { + errors.add(e); + } + executionLatch.countDown(); + }); + } + + assertThat(executionLatch.await(10, TimeUnit.SECONDS)).isTrue(); + synchronized (errors) { + assertThat(errors).isEmpty(); + } + + assertThat(clientInstances).hasValue(1); + + executorService.shutdown(); + assertThat(executorService.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + + sftpSessionFactory.destroy(); + } + } + }