CAMEL-23000: camel-ssh - idleTimeout

Author: Croway

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ssh.json

@@ -51,6 +51,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "healthCheckConsumerEnabled": target.setHealthCheckConsumerEnabled(property(camelContext, boolean.class, value)); return true;
         case "healthcheckproducerenabled":
         case "healthCheckProducerEnabled": target.setHealthCheckProducerEnabled(property(camelContext, boolean.class, value)); return true;
+        case "idletimeout":
+        case "idleTimeout": getOrCreateConfiguration(target).setIdleTimeout(property(camelContext, long.class, value)); return true;
         case "kex": getOrCreateConfiguration(target).setKex(property(camelContext, java.lang.String.class, value)); return true;
         case "keypairprovider":
         case "keyPairProvider": getOrCreateConfiguration(target).setKeyPairProvider(property(camelContext, org.apache.sshd.common.keyprovider.KeyPairProvider.class, value)); return true;
@@ -104,6 +106,8 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "healthCheckConsumerEnabled": return boolean.class;
         case "healthcheckproducerenabled":
         case "healthCheckProducerEnabled": return boolean.class;
+        case "idletimeout":
+        case "idleTimeout": return long.class;
         case "kex": return java.lang.String.class;
         case "keypairprovider":
         case "keyPairProvider": return org.apache.sshd.common.keyprovider.KeyPairProvider.class;
@@ -153,6 +157,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "healthCheckConsumerEnabled": return target.isHealthCheckConsumerEnabled();
         case "healthcheckproducerenabled":
         case "healthCheckProducerEnabled": return target.isHealthCheckProducerEnabled();
+        case "idletimeout":
+        case "idleTimeout": return getOrCreateConfiguration(target).getIdleTimeout();
         case "kex": return getOrCreateConfiguration(target).getKex();
         case "keypairprovider":
         case "keyPairProvider": return getOrCreateConfiguration(target).getKeyPairProvider();

components/camel-ssh/src/generated/java/org/apache/camel/component/ssh/SshComponentConfigurer.java

@@ -49,6 +49,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "failonunknownhost":
         case "failOnUnknownHost": target.getConfiguration().setFailOnUnknownHost(property(camelContext, boolean.class, value)); return true;
         case "greedy": target.setGreedy(property(camelContext, boolean.class, value)); return true;
+        case "idletimeout":
+        case "idleTimeout": target.getConfiguration().setIdleTimeout(property(camelContext, long.class, value)); return true;
         case "initialdelay":
         case "initialDelay": target.setInitialDelay(property(camelContext, long.class, value)); return true;
         case "kex": target.getConfiguration().setKex(property(camelContext, java.lang.String.class, value)); return true;
@@ -128,6 +130,8 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "failonunknownhost":
         case "failOnUnknownHost": return boolean.class;
         case "greedy": return boolean.class;
+        case "idletimeout":
+        case "idleTimeout": return long.class;
         case "initialdelay":
         case "initialDelay": return long.class;
         case "kex": return java.lang.String.class;
@@ -203,6 +207,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "failonunknownhost":
         case "failOnUnknownHost": return target.getConfiguration().isFailOnUnknownHost();
         case "greedy": return target.isGreedy();
+        case "idletimeout":
+        case "idleTimeout": return target.getConfiguration().getIdleTimeout();
         case "initialdelay":
         case "initialDelay": return target.getInitialDelay();
         case "kex": return target.getConfiguration().getKex();

components/camel-ssh/src/generated/java/org/apache/camel/component/ssh/SshEndpointConfigurer.java

@@ -23,7 +23,7 @@ public class SshEndpointUriFactory extends org.apache.camel.support.component.En
     private static final Set<String> SECRET_PROPERTY_NAMES;
     private static final Map<String, String> MULTI_VALUE_PREFIXES;
     static {
-        Set<String> props = new HashSet<>(41);
+        Set<String> props = new HashSet<>(42);
         props.add("backoffErrorThreshold");
         props.add("backoffIdleThreshold");
         props.add("backoffMultiplier");
@@ -40,6 +40,7 @@ public class SshEndpointUriFactory extends org.apache.camel.support.component.En
         props.add("failOnUnknownHost");
         props.add("greedy");
         props.add("host");
+        props.add("idleTimeout");
         props.add("initialDelay");
         props.add("kex");
         props.add("keyPairProvider");

components/camel-ssh/src/generated/java/org/apache/camel/component/ssh/SshEndpointUriFactory.java

@@ -74,6 +74,8 @@ public class SshConfiguration implements Cloneable {
     private String signatures;
     @UriParam(label = "advanced")
     private String compressions;
+    @UriParam(label = "advanced")
+    private long idleTimeout;
     @UriParam
     @Metadata(label = "advanced", autowired = true)
     private ClientBuilder clientBuilder;
@@ -396,6 +398,20 @@ public void setCompressions(String compressions) {
         this.compressions = compressions;
     }
 
+    public long getIdleTimeout() {
+        return idleTimeout;
+    }
+
+    /**
+     * Sets the timeout in milliseconds to wait before the SSH session is closed due to inactivity. The default value is
+     * 0, which means no idle timeout is applied.
+     *
+     * @param idleTimeout long milliseconds to wait before the session is closed due to inactivity.
+     */
+    public void setIdleTimeout(long idleTimeout) {
+        this.idleTimeout = idleTimeout;
+    }
+
     public ClientBuilder getClientBuilder() {
         return clientBuilder;
     }

components/camel-ssh/src/generated/resources/META-INF/org/apache/camel/component/ssh/ssh.json

@@ -21,8 +21,6 @@
 import org.apache.camel.support.ScheduledPollConsumer;
 import org.apache.sshd.client.SshClient;
 
-import static org.apache.camel.component.ssh.SshUtils.*;
-
 public class SshConsumer extends ScheduledPollConsumer {
     private final SshEndpoint endpoint;
 
@@ -35,15 +33,7 @@ public SshConsumer(SshEndpoint endpoint, Processor processor) {
 
     @Override
     protected void doStart() throws Exception {
-        if (this.endpoint.getConfiguration() == null || this.endpoint.getConfiguration().getClientBuilder() == null) {
-            client = SshClient.setUpDefaultClient();
-        } else {
-            client = this.endpoint.getConfiguration().getClientBuilder().build(true);
-        }
-        SshConfiguration configuration = endpoint.getConfiguration();
-        configureAlgorithms(configuration, client);
-
-        client.start();
+        client = SshUtils.createAndStartClient(endpoint.getConfiguration());
 
         super.doStart();
     }

components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshConfiguration.java

@@ -24,8 +24,6 @@
 import org.apache.camel.support.DefaultProducer;
 import org.apache.sshd.client.SshClient;
 
-import static org.apache.camel.component.ssh.SshUtils.*;
-
 public class SshProducer extends DefaultProducer {
     private SshEndpoint endpoint;
     private SshClient client;
@@ -37,14 +35,7 @@ public SshProducer(SshEndpoint endpoint) {
 
     @Override
     protected void doStart() throws Exception {
-        if (this.endpoint.getConfiguration() == null || this.endpoint.getConfiguration().getClientBuilder() == null) {
-            client = SshClient.setUpDefaultClient();
-        } else {
-            client = this.endpoint.getConfiguration().getClientBuilder().build(true);
-        }
-        SshConfiguration configuration = endpoint.getConfiguration();
-        configureAlgorithms(configuration, client);
-        client.start();
+        client = SshUtils.createAndStartClient(endpoint.getConfiguration());
 
         super.doStart();
     }

components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshConsumer.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.ssh;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
@@ -43,6 +44,7 @@
 import org.apache.sshd.common.signature.BuiltinSignatures;
 import org.apache.sshd.common.signature.Signature;
 import org.apache.sshd.common.signature.SignatureFactory;
+import org.apache.sshd.core.CoreModuleProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -168,4 +170,19 @@ public static void configureAlgorithms(SshConfiguration configuration, SshClient
         configureCompressions(configuration.getCompressions(), client);
     }
 
+    public static SshClient createAndStartClient(SshConfiguration configuration) throws Exception {
+        SshClient client;
+        if (configuration == null || configuration.getClientBuilder() == null) {
+            client = SshClient.setUpDefaultClient();
+        } else {
+            client = configuration.getClientBuilder().build(true);
+        }
+        configureAlgorithms(configuration, client);
+        if (configuration.getIdleTimeout() > 0) {
+            CoreModuleProperties.IDLE_TIMEOUT.set(client, Duration.ofMillis(configuration.getIdleTimeout()));
+        }
+        client.start();
+        return client;
+    }
+
 }

components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshProducer.java

@@ -0,0 +1,112 @@
+/*
+ * 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.camel.component.ssh;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.sshd.server.Environment;
+import org.apache.sshd.server.ExitCallback;
+import org.apache.sshd.server.channel.ChannelSession;
+import org.apache.sshd.server.command.Command;
+import org.apache.sshd.server.command.CommandFactory;
+
+/**
+ * A command factory that introduces a delay before echoing the command back. Used to test idle timeout behavior — if
+ * the client's idle timeout is shorter than the delay, the session will be closed before the command completes.
+ */
+public class DelayedEchoCommandFactory implements CommandFactory {
+
+    private final long delayMs;
+
+    public DelayedEchoCommandFactory(long delayMs) {
+        this.delayMs = delayMs;
+    }
+
+    @Override
+    public Command createCommand(ChannelSession channelSession, String command) {
+        return new DelayedEchoCommand(command, delayMs);
+    }
+
+    protected static class DelayedEchoCommand implements Command, Runnable {
+        private final String command;
+        private final long delayMs;
+        private OutputStream out;
+        private OutputStream err;
+        private ExitCallback callback;
+        private Thread thread;
+
+        public DelayedEchoCommand(String command, long delayMs) {
+            this.command = command;
+            this.delayMs = delayMs;
+        }
+
+        @Override
+        public void setInputStream(InputStream in) {
+        }
+
+        @Override
+        public void setOutputStream(OutputStream out) {
+            this.out = out;
+        }
+
+        @Override
+        public void setErrorStream(OutputStream err) {
+            this.err = err;
+        }
+
+        @Override
+        public void setExitCallback(ExitCallback callback) {
+            this.callback = callback;
+        }
+
+        @Override
+        public void start(ChannelSession channelSession, Environment environment) throws IOException {
+            thread = new Thread(this, "DelayedEchoCommand");
+            thread.start();
+        }
+
+        @Override
+        public void destroy(ChannelSession channelSession) throws Exception {
+            // noop
+        }
+
+        @Override
+        public void run() {
+            boolean succeeded = true;
+            String message = null;
+            try {
+                Thread.sleep(delayMs);
+                err.write("Error:".getBytes());
+                err.write(command.getBytes());
+                err.flush();
+                out.write(command.getBytes());
+                out.flush();
+            } catch (Exception e) {
+                succeeded = false;
+                message = e.toString();
+            } finally {
+                if (succeeded) {
+                    callback.onExit(0);
+                } else {
+                    callback.onExit(1, message);
+                }
+            }
+        }
+    }
+}