Support idle timeout

src/main/java/com/rabbitmq/model/ConnectionSettings.java

@@ -17,6 +17,7 @@
 // [email protected].
 package com.rabbitmq.model;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.function.Function;
 
@@ -38,5 +39,7 @@ public interface ConnectionSettings<T> {
 
   T credentialsProvider(CredentialsProvider credentialsProvider);
 
+  T idleTimeout(Duration idleTimeout);
+
   T addressSelector(Function<List<Address>, Address> selector);
 }

src/main/java/com/rabbitmq/model/amqp/AmqpConnection.java

@@ -184,6 +184,8 @@ private org.apache.qpid.protonj2.client.Connection connect(
     connectionOptions.virtualHost("vhost:" + connectionSettings.virtualHost());
     // only the mechanisms supported in RabbitMQ
     connectionOptions.saslOptions().addAllowedMechanism("PLAIN").addAllowedMechanism("EXTERNAL");
+    connectionOptions.idleTimeout(
+        connectionSettings.idleTimeout().toMillis(), TimeUnit.MILLISECONDS);
     connectionOptions.disconnectedHandler(disconnectHandler);
     if (name != null) {
       connectionOptions.properties(singletonMap("connection_name", name));

src/main/java/com/rabbitmq/model/amqp/AmqpConnectionBuilder.java

@@ -79,6 +79,11 @@ public ConnectionBuilder credentialsProvider(CredentialsProvider credentialsProv
     return this.connectionSettings.credentialsProvider(credentialsProvider);
   }
 
+  @Override
+  public ConnectionBuilder idleTimeout(Duration idleTimeout) {
+    return this.connectionSettings.idleTimeout(idleTimeout);
+  }
+
   @Override
   public ConnectionBuilder addressSelector(Function<List<Address>, Address> selector) {
     return this.connectionSettings.addressSelector(selector);

src/main/java/com/rabbitmq/model/amqp/DefaultConnectionSettings.java

@@ -25,11 +25,13 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URLDecoder;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
+import org.apache.qpid.protonj2.client.ConnectionOptions;
 
 abstract class DefaultConnectionSettings<T> implements ConnectionSettings<T> {
 
@@ -46,6 +48,7 @@ void copyTo(ConnectionSettings<?> copy) {
     copy.virtualHost(this.virtualHost);
     copy.uris(this.uris.stream().map(URI::toString).toArray(String[]::new));
     copy.addressSelector(this.addressSelector);
+    copy.idleTimeout(this.idleTimeout);
   }
 
   private String host = DEFAULT_HOST;
@@ -54,6 +57,7 @@ void copyTo(ConnectionSettings<?> copy) {
       new DefaultUsernamePasswordCredentialsProvider(DEFAULT_USERNAME, DEFAULT_PASSWORD);
   private String virtualHost = DEFAULT_VIRTUAL_HOST;
   private List<URI> uris = Collections.emptyList();
+  private Duration idleTimeout = Duration.ofMillis(ConnectionOptions.DEFAULT_IDLE_TIMEOUT);
   private static final Random RANDOM = new Random();
   private Function<List<Address>, Address> addressSelector =
       addresses -> {
@@ -131,6 +135,15 @@ public T credentialsProvider(CredentialsProvider credentialsProvider) {
     return this.toReturn();
   }
 
+  @Override
+  public T idleTimeout(Duration idleTimeout) {
+    if (idleTimeout.isNegative()) {
+      throw new IllegalArgumentException("Idle timeout cannot be negative");
+    }
+    this.idleTimeout = idleTimeout;
+    return this.toReturn();
+  }
+
   @Override
   public T addressSelector(Function<List<Address>, Address> selector) {
     this.addressSelector = selector;
@@ -145,6 +158,10 @@ String virtualHost() {
     return virtualHost;
   }
 
+  Duration idleTimeout() {
+    return idleTimeout;
+  }
+
   Address selectAddress() {
     return this.addressSelector.apply(this.addresses);
   }