SftpFileSystem: keep a session until the filesystem is closed
When an SftpFileSystem is obtained via FileSystems.newFileSystem(URI),
the new file system is created with an implicit SshClient and a new
implicit ClientSession. This session has a heartbeat set to keep it
open for the lifetime of the file system, i.e., until the file system
is closed.
But heartbeats do not guarantee that the session remains open. A server
may decide all the same to close the session, for instance, if it is
idle for too long. (Even despite heartbeat messages.)
So ensure that in such a case, the file system automatically tries to
open a new session. Otherwise a client of such an SftpFileSystem must
be prepared to deal with these cases at any time, which may be difficult
or even impossible to do.
diff --git a/docs/technical/sftp_filesystem.md b/docs/technical/sftp_filesystem.md
index 36e197d..c814d17 100644
--- a/docs/technical/sftp_filesystem.md
+++ b/docs/technical/sftp_filesystem.md
@@ -17,15 +17,16 @@
off that session using `SftpClientFactory.instance().createSftpFileSystem()`.
The file system remains valid until it is closed, or until the session is
closed. When the file system is closed, the session will *not* be closed.
+ When the session closes, so will the file system.
2. You can create an `SftpFileSystem` with a `sftp://` URI using the standard
Java factory `java.nio.file.FileSystems.newFileSystem()`. This will automatically
create an `SshClient` with default settings, and the file system will open
an SSH session itself. This session has heartbeats enabled to keep it open
for as long as the file system is open. The file system remains valid until
- closed, at which point it will close the session it had created.
-
-In either case, the file system will be closed if the session closes.
+ closed, at which point it will close the session it had created. If the
+ session closes while the file system is not closed, the next operation on
+ such a file system will open a new session.
# SSH Resource Management
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystem.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystem.java
index a49fa44..a4bd87b 100644
--- a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystem.java
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystem.java
@@ -86,53 +86,68 @@
public static final AttributeKey<Boolean> OWNED_SESSION = new AttributeKey<>();
private final String id;
- private final ClientSession clientSession;
private final SftpClientFactory factory;
private final SftpVersionSelector selector;
private final SftpErrorDataHandler errorDataHandler;
- private final SftpClientPool pool;
- private final int version;
- private final Set<String> supportedViews;
+ private SftpClientPool pool;
+ private int version;
+ private Set<String> supportedViews;
private SftpPath defaultDir;
private int readBufferSize;
private int writeBufferSize;
private final List<FileStore> stores;
private final AtomicBoolean open = new AtomicBoolean();
+ private AtomicReference<ClientSession> clientSession = new AtomicReference<>();
+
public SftpFileSystem(SftpFileSystemProvider provider, String id, ClientSession session,
SftpClientFactory factory, SftpVersionSelector selector, SftpErrorDataHandler errorDataHandler)
throws IOException {
- super(provider);
- this.id = id;
- this.clientSession = Objects.requireNonNull(session, "No client session");
- clientSession.addSessionListener(new SessionListener() {
+ this(provider, id, factory, selector, errorDataHandler);
+ clientSession.set(Objects.requireNonNull(session, "No client session"));
+ if (!Boolean.TRUE.equals(session.getAttribute(OWNED_SESSION))) {
+ session.addSessionListener(new SessionListener() {
- @Override
- public void sessionClosed(Session session) {
- if (clientSession == session) {
- try {
- close();
- } catch (IOException e) {
- log.warn("sessionClosed({}) [{}] could not close file system properly: {}", session,
- SftpFileSystem.this, e.toString(), e);
+ @Override
+ public void sessionClosed(Session session) {
+ if (clientSession.get() == session) {
+ try {
+ close();
+ } catch (IOException e) {
+ log.warn("sessionClosed({}) [{}] could not close file system properly: {}", session,
+ SftpFileSystem.this, e.toString(), e);
+ }
}
}
- }
- });
+ });
+ }
+ init();
+ }
+
+ protected SftpFileSystem(SftpFileSystemProvider provider, String id, SftpClientFactory factory,
+ SftpVersionSelector selector, SftpErrorDataHandler errorDataHandler) {
+ super(provider);
+ this.id = id;
this.factory = factory != null ? factory : SftpClientFactory.instance();
this.selector = selector;
this.errorDataHandler = errorDataHandler;
- this.stores = Collections.unmodifiableList(Collections.<FileStore> singletonList(new SftpFileStore(id, this)));
- this.pool = new SftpClientPool(SftpModuleProperties.POOL_SIZE.getRequired(session),
- SftpModuleProperties.POOL_LIFE_TIME.getRequired(session),
- SftpModuleProperties.POOL_CORE_SIZE.getRequired(session));
+ this.stores = Collections.singletonList(new SftpFileStore(id, this));
+ }
+
+ protected void init() throws IOException {
open.set(true);
try (SftpClient client = getClient()) {
+ ClientSession session = client.getClientSession();
+ pool = new SftpClientPool(SftpModuleProperties.POOL_SIZE.getRequired(session),
+ SftpModuleProperties.POOL_LIFE_TIME.getRequired(session),
+ SftpModuleProperties.POOL_CORE_SIZE.getRequired(session));
version = client.getVersion();
defaultDir = getPath(client.canonicalPath("."));
} catch (RuntimeException | IOException e) {
open.set(false);
- pool.close();
+ if (pool != null) {
+ pool.close();
+ }
throw e;
}
@@ -205,7 +220,7 @@
@Override
public ClientSession getClientSession() {
- return clientSession;
+ return clientSession.get();
}
@Override
@@ -213,6 +228,14 @@
return getClientSession();
}
+ protected void setClientSession(ClientSession newSession) {
+ clientSession.set(newSession);
+ }
+
+ protected ClientSession sessionForSftpClient() throws IOException {
+ return getClientSession();
+ }
+
@SuppressWarnings("synthetic-access")
public SftpClient getClient() throws IOException {
Wrapper wrapper = null;
@@ -220,9 +243,9 @@
if (!isOpen()) {
throw new IOException("SftpFileSystem is closed" + this);
}
- SftpClient client = pool.poll();
+ SftpClient client = pool != null ? pool.poll() : null;
if (client == null) {
- ClientSession session = getClientSession();
+ ClientSession session = sessionForSftpClient();
client = factory.createSftpClient(session, getSftpVersionSelector(), getSftpErrorDataHandler());
}
if (!client.isClosing()) {
@@ -235,7 +258,9 @@
@Override
public void close() throws IOException {
if (open.getAndSet(false)) {
- pool.close();
+ if (pool != null) {
+ pool.close();
+ }
SftpFileSystemProvider provider = provider();
String fsId = getId();
SftpFileSystem fs = provider.removeFileSystem(fsId);
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemAutomatic.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemAutomatic.java
new file mode 100644
index 0000000..763f980
--- /dev/null
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemAutomatic.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sshd.sftp.client.fs;
+
+import java.io.IOException;
+
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.util.io.functors.IOFunction;
+import org.apache.sshd.sftp.client.SftpClientFactory;
+import org.apache.sshd.sftp.client.SftpErrorDataHandler;
+import org.apache.sshd.sftp.client.SftpVersionSelector;
+
+/**
+ * An {@link SftpFileSystem} that uses a provider function for its {@link ClientSession} so that it can continue to
+ * function even if a session was closed. The provider is supposed to create a new session if the current one is not
+ * open.
+ *
+ * @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
+ */
+public class SftpFileSystemAutomatic extends SftpFileSystem {
+
+ private final IOFunction<Boolean, ClientSession> sessionProvider;
+
+ public SftpFileSystemAutomatic(SftpFileSystemProvider provider, String id,
+ IOFunction<Boolean, ClientSession> sessionProvider, SftpClientFactory factory,
+ SftpVersionSelector selector, SftpErrorDataHandler errorDataHandler)
+ throws IOException {
+ super(provider, id, factory, selector, errorDataHandler);
+ this.sessionProvider = sessionProvider;
+ init();
+ }
+
+ @Override
+ public ClientSession getClientSession() {
+ try {
+ return sessionProvider.apply(Boolean.FALSE);
+ } catch (IOException e) {
+ // Cannot occur
+ return null;
+ }
+ }
+
+ @Override
+ protected ClientSession sessionForSftpClient() throws IOException {
+ if (!isOpen()) {
+ throw new IOException("SftpFileSystem is closed" + this);
+ }
+ ClientSession result = sessionProvider.apply(Boolean.TRUE);
+ setClientSession(result);
+ return result;
+ }
+
+}
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemClientSessionInitializer.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemClientSessionInitializer.java
index 890864e..c7e70d7 100644
--- a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemClientSessionInitializer.java
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemClientSessionInitializer.java
@@ -20,17 +20,20 @@
package org.apache.sshd.sftp.client.fs;
import java.io.IOException;
+import java.time.Duration;
import java.util.Map;
import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.client.session.ClientSessionCreator;
import org.apache.sshd.common.auth.PasswordHolder;
import org.apache.sshd.common.auth.UsernameHolder;
+import org.apache.sshd.common.session.SessionHeartbeatController;
+import org.apache.sshd.common.util.io.functors.IOFunction;
import org.apache.sshd.sftp.client.SftpErrorDataHandler;
import org.apache.sshd.sftp.client.SftpVersionSelector;
/**
- * Provides user hooks into the process of creating a {@link SftpFileSystem} via a {@link SftpFileSystemProvider}
+ * Provides user hooks into the process of creating a {@link SftpFileSystem} via a {@link SftpFileSystemProvider}.
*
* @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
*/
@@ -81,6 +84,32 @@
session.addPasswordIdentity(password);
}
session.auth().verify(context.getMaxAuthTime());
+
+ // Keep this session open for as long as the filesystem is in use.
+ // TODO: the interval should be configurable somehow.
+ session.setSessionHeartbeat(SessionHeartbeatController.HeartbeatType.IGNORE, Duration.ofSeconds(10));
+ }
+
+ /**
+ * Invoked by the {@link SftpFileSystemProvider#newFileSystem(java.net.URI, Map)} method in order to create the
+ * {@link SftpFileSystem} once session has been authenticated.
+ *
+ * @param provider The {@link SftpFileSystemProvider} instance requesting the session
+ * @param context The initialization {@link SftpFileSystemInitializationContext}
+ * @param session The authenticated {@link ClientSession}
+ * @param selector The <U>resolved</U> {@link SftpVersionSelector} to use
+ * @param errorDataHandler The {@link SftpErrorDataHandler} to handle incoming data through the error stream -
+ * if {@code null} the data is silently ignored
+ * @return The created {@link SftpFileSystem}
+ * @throws IOException If failed to create the file-system
+ * @deprecated since 2.10.1
+ */
+ @Deprecated
+ default SftpFileSystem createSftpFileSystem(
+ SftpFileSystemProvider provider, SftpFileSystemInitializationContext context, ClientSession session,
+ SftpVersionSelector selector, SftpErrorDataHandler errorDataHandler)
+ throws IOException {
+ return createSftpFileSystem(provider, context, x -> session, selector, errorDataHandler);
}
/**
@@ -89,7 +118,7 @@
*
* @param provider The {@link SftpFileSystemProvider} instance requesting the session
* @param context The initialization {@link SftpFileSystemInitializationContext}
- * @param session The authenticated {@link ClientSession}
+ * @param sessionProvider A factory for authenticated {@link ClientSession}s
* @param selector The <U>resolved</U> {@link SftpVersionSelector} to use
* @param errorDataHandler The {@link SftpErrorDataHandler} to handle incoming data through the error stream - if
* {@code null} the data is silently ignored
@@ -97,10 +126,10 @@
* @throws IOException If failed to create the file-system
*/
default SftpFileSystem createSftpFileSystem(
- SftpFileSystemProvider provider, SftpFileSystemInitializationContext context, ClientSession session,
- SftpVersionSelector selector, SftpErrorDataHandler errorDataHandler)
- throws IOException {
- return new SftpFileSystem(
- provider, context.getId(), session, provider.getSftpClientFactory(), selector, errorDataHandler);
+ SftpFileSystemProvider provider, SftpFileSystemInitializationContext context,
+ IOFunction<Boolean, ClientSession> sessionProvider, SftpVersionSelector selector,
+ SftpErrorDataHandler errorDataHandler) throws IOException {
+ return new SftpFileSystemAutomatic(provider, context.getId(), sessionProvider, provider.getSftpClientFactory(),
+ selector, errorDataHandler);
}
}
diff --git a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemProvider.java b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemProvider.java
index c0c937d..06f5f30 100644
--- a/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemProvider.java
+++ b/sshd-sftp/src/main/java/org/apache/sshd/sftp/client/fs/SftpFileSystemProvider.java
@@ -71,6 +71,7 @@
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.session.ClientSession;
@@ -82,11 +83,14 @@
import org.apache.sshd.common.auth.BasicCredentialsProvider;
import org.apache.sshd.common.auth.MutableBasicCredentials;
import org.apache.sshd.common.io.IoSession;
+import org.apache.sshd.common.session.Session;
+import org.apache.sshd.common.session.SessionListener;
import org.apache.sshd.common.util.GenericUtils;
import org.apache.sshd.common.util.MapEntryUtils;
import org.apache.sshd.common.util.NumberUtils;
import org.apache.sshd.common.util.ValidateUtils;
import org.apache.sshd.common.util.io.IoUtils;
+import org.apache.sshd.common.util.io.functors.IOFunction;
import org.apache.sshd.common.util.logging.LoggingUtils;
import org.apache.sshd.sftp.SftpModuleProperties;
import org.apache.sshd.sftp.client.SftpClient;
@@ -246,48 +250,24 @@
Charset decodingCharset = SftpModuleProperties.NAME_DECODER_CHARSET.getRequired(resolver);
SftpFileSystemClientSessionInitializer initializer = getSftpFileSystemClientSessionInitializer();
- SftpFileSystem fileSystem;
+ SftpFileSystem fileSystem = null;
synchronized (fileSystems) {
if (fileSystems.containsKey(id)) {
throw new FileSystemAlreadyExistsException(id);
}
- // TODO try and find a way to avoid doing this while locking the file systems cache
- ClientSession session = null;
+ SessionProvider sessionProvider = new SessionProvider(context, params, decodingCharset);
try {
- session = initializer.createClientSession(this, context);
-
- // Make any extra configuration parameters available to the session
- if (MapEntryUtils.size(params) > 0) {
- // Cannot use forEach because the session is not effectively final
- for (Map.Entry<String, ?> pe : params.entrySet()) {
- String key = pe.getKey();
- Object value = pe.getValue();
- if (VERSION_PARAM.equalsIgnoreCase(key)) {
- continue;
- }
-
- PropertyResolverUtils.updateProperty(session, key, value);
- }
-
- SftpModuleProperties.NAME_DECODING_CHARSET.set(session, decodingCharset);
- }
-
- initializer.authenticateClientSession(this, context, session);
-
- session.setAttribute(SftpFileSystem.OWNED_SESSION, Boolean.TRUE);
-
- fileSystem = initializer.createSftpFileSystem(
- this, context, session, selector, errorHandler);
+ fileSystem = initializer.createSftpFileSystem(this, context, sessionProvider, selector, errorHandler);
fileSystems.put(id, fileSystem);
} catch (Exception e) {
- if (session != null) {
+ if (fileSystem != null) {
try {
- session.close();
+ fileSystem.close();
} catch (IOException t) {
e.addSuppressed(t);
LoggingUtils.debug(log,
- "Failed ({}) to close session for new file system on {}}:{} due to {}[{}]: {}",
+ "Failed ({}) to close new failed file system on {}}:{} due to {}[{}]: {}",
t.getClass().getSimpleName(), host, port, e.getClass().getSimpleName(), e.getMessage(),
t.getMessage(),
t);
@@ -318,6 +298,97 @@
return fileSystem;
}
+ /**
+ * A session provider that automatically creates a new session if the current one is no longer open (or if there
+ * isn't one yet). It returns fully authenticated sessions.
+ */
+ private class SessionProvider implements IOFunction<Boolean, ClientSession> {
+
+ private final SftpFileSystemInitializationContext context;
+
+ private final Map<String, ?> params;
+
+ private final Charset decodingCharset;
+
+ private AtomicReference<ClientSession> currentSession = new AtomicReference<>();
+
+ SessionProvider(SftpFileSystemInitializationContext context, Map<String, ?> params, Charset decodingCharset) {
+ this.context = Objects.requireNonNull(context);
+ this.params = Objects.requireNonNull(params);
+ this.decodingCharset = Objects.requireNonNull(decodingCharset);
+ }
+
+ /**
+ * Retrieves the current {@link ClientSession} and optionally creates a new one if there is no current session
+ * yet, or if it is not open.
+ *
+ * @param create {@link Boolean#TRUE} to create a new session if needed, otherwise just query the current
+ * session.
+ */
+ @Override
+ public ClientSession apply(Boolean create) throws IOException {
+ synchronized (this) {
+ ClientSession session = currentSession.get();
+ if ((session == null || !session.isOpen()) && Boolean.TRUE.equals(create)) {
+ session = create();
+ currentSession.set(session);
+ }
+ return session;
+ }
+ }
+
+ private ClientSession create() throws IOException {
+ SftpFileSystemClientSessionInitializer initializer = getSftpFileSystemClientSessionInitializer();
+ ClientSession session = null;
+ try {
+ session = initializer.createClientSession(SftpFileSystemProvider.this, context);
+ ClientSession mySelf = session;
+
+ // Make any extra configuration parameters available to the session
+ params.forEach((key, value) -> {
+ if (!VERSION_PARAM.equalsIgnoreCase(key)) {
+ PropertyResolverUtils.updateProperty(mySelf, key, value);
+ }
+ });
+ SftpModuleProperties.NAME_DECODING_CHARSET.set(session, decodingCharset);
+
+ initializer.authenticateClientSession(SftpFileSystemProvider.this, context, session);
+
+ session.setAttribute(SftpFileSystem.OWNED_SESSION, Boolean.TRUE);
+ session.addSessionListener(new SessionListener() {
+
+ @Override
+ public void sessionClosed(Session s) {
+ if (mySelf == s) {
+ currentSession.compareAndSet(mySelf, null);
+ }
+ }
+ });
+ return session;
+ } catch (Exception e) {
+ if (session != null) {
+ try {
+ session.close();
+ } catch (IOException t) {
+ e.addSuppressed(t);
+ LoggingUtils.debug(log, "Failed ({}) to close session for new file system on {}}:{} due to {}[{}]: {}",
+ t.getClass().getSimpleName(), context.getHost(), context.getPort(),
+ e.getClass().getSimpleName(), e.getMessage(), t.getMessage(), t);
+ }
+ }
+
+ if (e instanceof IOException) {
+ throw (IOException) e;
+ } else if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ } else {
+ throw new IOException(e);
+ }
+ }
+ }
+
+ }
+
protected SftpVersionSelector resolveSftpVersionSelector(
URI uri, SftpVersionSelector defaultSelector, PropertyResolver resolver) {
String preference = resolver.getString(VERSION_PARAM);
diff --git a/sshd-sftp/src/test/java/org/apache/sshd/sftp/client/fs/SftpFileSystemTest.java b/sshd-sftp/src/test/java/org/apache/sshd/sftp/client/fs/SftpFileSystemTest.java
index b92df1c..82d9876 100644
--- a/sshd-sftp/src/test/java/org/apache/sshd/sftp/client/fs/SftpFileSystemTest.java
+++ b/sshd-sftp/src/test/java/org/apache/sshd/sftp/client/fs/SftpFileSystemTest.java
@@ -861,6 +861,42 @@
}
@Test
+ public void testSessionRecreate() throws Exception {
+ Path targetPath = detectTargetFolder();
+ Path lclSftp = CommonTestSupportUtils.resolve(targetPath, SftpConstants.SFTP_SUBSYSTEM_NAME,
+ getClass().getSimpleName());
+ Files.createDirectories(lclSftp);
+
+ Path lclFile = lclSftp.resolve(getCurrentTestName() + ".txt");
+ Files.deleteIfExists(lclFile);
+ byte[] expected = (getClass().getName() + "#" + getCurrentTestName() + "(" + new Date() + ")")
+ .getBytes(StandardCharsets.UTF_8);
+ ClientSession session;
+ FileSystem fs = FileSystems.newFileSystem(createDefaultFileSystemURI(), Collections.emptyMap());
+ try {
+ assertTrue("Should be an SftpFileSystem", fs instanceof SftpFileSystem);
+ Path remotePath = fs.getPath(CommonTestSupportUtils.resolveRelativeRemotePath(targetPath.getParent(), lclFile));
+ Files.write(remotePath, "foo".getBytes(StandardCharsets.US_ASCII));
+ session = ((SftpFileSystem) fs).getClientSession();
+ session.close();
+ Files.write(remotePath, expected);
+ ClientSession session2 = ((SftpFileSystem) fs).getClientSession();
+ assertNotNull("Expected a session", session2);
+ assertNotSame("Expected different sessions", session, session2);
+ session = session2;
+ assertTrue("Second session should still be open", session.isOpen());
+ assertTrue("Recreated session should be owned by the file system",
+ session.getAttribute(SftpFileSystem.OWNED_SESSION));
+ } finally {
+ fs.close();
+ }
+ byte[] actual = Files.readAllBytes(lclFile);
+ assertArrayEquals("Mismatched persisted data", expected, actual);
+ assertFalse("File system should not be open", fs.isOpen());
+ assertFalse("Owned session should not be open", session.isOpen());
+ }
+
+ @Test
public void testFileSystemProviderServiceEntry() throws IOException {
Path configFile = CommonTestSupportUtils.resolve(detectSourcesFolder(),
MAIN_SUBFOLDER, "filtered-resources", "META-INF", "services", FileSystemProvider.class.getName());
