Adjust expiration time for OAuth tests

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/OAuthSettings.java

@@ -19,14 +19,53 @@
 
 import javax.net.ssl.SSLContext;
 
+/**
+ * Configuration to retrieve a token using the <a
+ * href="https://tools.ietf.org/html/rfc6749#section-4.4">OAuth 2 Client Credentials flow</a>.
+ *
+ * @param <T> the type of object returned by methods, usually the object itself
+ */
 public interface OAuthSettings<T> {
 
+  /**
+   * Set the URI to access to get the token.
+   *
+   * <p>TLS is supported by providing a <code>HTTPS</code> URI and setting a {@link SSLContext}. See
+   * {@link #tls()} for more information. <em>Applications in production should always use HTTPS to
+   * retrieve tokens.</em>
+   *
+   * @param uri access URI
+   * @return OAuth settings
+   */
   OAuthSettings<T> tokenEndpointUri(String uri);
 
+  /**
+   * Set the OAuth 2 client ID
+   *
+   * <p>The client ID usually identifies the application that requests a token.
+   *
+   * @param clientId client ID
+   * @return OAuth settings
+   */
   OAuthSettings<T> clientId(String clientId);
 
+  /**
+   * Set the secret (password) to use to get a token.
+   *
+   * @param clientSecret client secret
+   * @return OAuth settings
+   */
   OAuthSettings<T> clientSecret(String clientSecret);
 
+  /**
+   * Set the grant type to use when requesting the token.
+   *
+   * <p>The default is <code>client_credentials</code>, but some OAuth 2 servers can use
+   * non-standard grant types to request tokens with extra-information.
+   *
+   * @param grantType grant type
+   * @return OAuth settings
+   */
   OAuthSettings<T> grantType(String grantType);
 
   OAuthSettings<T> parameter(String name, String value);

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -18,13 +18,15 @@
 package com.rabbitmq.client.amqp.impl;
 
 import static com.rabbitmq.client.amqp.Resource.State.*;
+import static com.rabbitmq.client.amqp.impl.ExceptionUtils.convert;
 import static com.rabbitmq.client.amqp.impl.Utils.supportFilterExpressions;
 import static com.rabbitmq.client.amqp.impl.Utils.supportSetToken;
 import static java.lang.System.nanoTime;
 import static java.time.Duration.ofNanos;
 
 import com.rabbitmq.client.amqp.*;
 import com.rabbitmq.client.amqp.ObservationCollector;
+import com.rabbitmq.client.amqp.impl.Utils.RunnableWithException;
 import com.rabbitmq.client.amqp.impl.Utils.StopWatch;
 import com.rabbitmq.client.amqp.metrics.MetricsCollector;
 import java.time.Duration;
@@ -102,8 +104,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
     } else {
       disconnectHandler =
           (c, e) -> {
-            AmqpException failureCause =
-                ExceptionUtils.convert(e.failureCause(), "Connection disconnected");
+            AmqpException failureCause = convert(e.failureCause(), "Connection disconnected");
             this.close(failureCause);
           };
     }
@@ -248,7 +249,7 @@ private NativeConnectionWrapper connect(
       checkBrokerVersion(connection);
       return new NativeConnectionWrapper(connection, extractNode(connection), address);
     } catch (ClientException e) {
-      throw ExceptionUtils.convert(e);
+      throw convert(e);
     } finally {
       LOGGER.debug("Connection attempt for '{}' took {}", this.name(), stopWatch.stop());
     }
@@ -275,7 +276,7 @@ private static String brokerVersion(org.apache.qpid.protonj2.client.Connection c
     try {
       return (String) connection.properties().get("version");
     } catch (ClientException e) {
-      throw ExceptionUtils.convert(e);
+      throw convert(e);
     }
   }
 
@@ -315,11 +316,11 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
         resultReference = new AtomicReference<>();
     BiConsumer<org.apache.qpid.protonj2.client.Connection, DisconnectionEvent> result =
         (conn, event) -> {
-          ClientIOException ioex = event.failureCause();
+          ClientIOException failureCause = event.failureCause();
           LOGGER.debug(
               "Disconnect handler of '{}', error is the following: {}",
               this.name(),
-              ioex.getMessage());
+              failureCause.getMessage());
           if (this.state() == OPENING) {
             LOGGER.debug("Connection is still opening, disconnect handler skipped");
             // the broker is not available when opening the connection
@@ -332,7 +333,8 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
                 this.name());
             return;
           }
-          AmqpException exception = ExceptionUtils.convert(event.failureCause());
+          AmqpException exception =
+              convert(failureCause, "Connection '%s' disconnected", this.name());
           LOGGER.debug("Converted native exception to {}", exception.getClass().getSimpleName());
 
           if (RECOVERY_PREDICATE.test(exception) && this.state() != OPENING) {
@@ -353,8 +355,8 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
             LOGGER.debug(
                 "Not recovering connection '{}' for error {}",
                 this.name(),
-                event.failureCause().getMessage());
-            close(ExceptionUtils.convert(ioex));
+                failureCause.getMessage());
+            close(exception);
           }
         };
 
@@ -596,7 +598,7 @@ private Session openSession(org.apache.qpid.protonj2.client.Connection connectio
     try {
       return connection.openSession();
     } catch (ClientException e) {
-      throw ExceptionUtils.convert(e, "Error while opening session");
+      throw convert(e, "Error while opening session");
     }
   }
 
@@ -745,48 +747,47 @@ long id() {
   private void close(Throwable cause) {
     if (this.closed.compareAndSet(false, true)) {
       this.state(CLOSING, cause);
+      LOGGER.debug("Closing connection {}", this);
       this.credentialsRegistration.unregister();
       this.environment.removeConnection(this);
+      BiConsumer<String, RunnableWithException> safeClose =
+          (label, action) -> {
+            try {
+              action.run();
+            } catch (Exception e) {
+              LOGGER.info(
+                  "Error during connection '{}' closing ({}): {}", this, label, e.getMessage());
+            }
+          };
       if (this.topologyListener instanceof AutoCloseable) {
-        try {
-          ((AutoCloseable) this.topologyListener).close();
-        } catch (Exception e) {
-          LOGGER.info("Error while closing topology listener", e);
-        }
+        safeClose.accept(
+            "topology listener", () -> ((AutoCloseable) this.topologyListener).close());
       }
-      this.closeManagement();
+      safeClose.accept("management", this::closeManagement);
+
       for (RpcClient rpcClient : this.rpcClients) {
-        rpcClient.close();
+        safeClose.accept("RPC client", rpcClient::close);
       }
       for (RpcServer rpcServer : this.rpcServers) {
-        rpcServer.close();
+        safeClose.accept("RPC server", rpcServer::close);
       }
       for (AmqpPublisher publisher : this.publishers) {
-        publisher.close(cause);
+        safeClose.accept("publisher", () -> publisher.close(cause));
       }
       for (AmqpConsumer consumer : this.consumers) {
-        consumer.close(cause);
+        safeClose.accept("consumer", () -> consumer.close(cause));
       }
-      try {
-        if (this.dispatchingExecutorService != null) {
-          this.dispatchingExecutorService.shutdownNow();
-        }
-      } catch (Exception e) {
-        LOGGER.info(
-            "Error while shutting down dispatching executor service for connection '{}': {}",
-            this.name(),
-            e.getMessage());
+      if (this.dispatchingExecutorService != null) {
+        safeClose.accept(
+            "dispatcing executor service", () -> this.dispatchingExecutorService.shutdown());
       }
-      try {
-        org.apache.qpid.protonj2.client.Connection nc = this.nativeConnection;
-        if (nc != null) {
-          nc.close();
-        }
-      } catch (Exception e) {
-        LOGGER.warn("Error while closing native connection", e);
+      org.apache.qpid.protonj2.client.Connection nc = this.nativeConnection;
+      if (nc != null) {
+        safeClose.accept("native connection", nc::close);
       }
       this.state(CLOSED, cause);
       this.environment.metricsCollector().closeConnection();
+      LOGGER.debug("Connection {} has been closed", this);
     }
   }
 

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -55,7 +55,6 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
   private volatile ClientReceiver nativeReceiver;
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final int initialCredits;
-  private final MessageHandler messageHandler;
   private final Long id;
   private final String address;
   private final String queue;
@@ -84,7 +83,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
     super(builder.listeners());
     this.id = ID_SEQUENCE.getAndIncrement();
     this.initialCredits = builder.initialCredits();
-    this.messageHandler =
+    MessageHandler messageHandler =
         builder
             .connection()
             .observationCollector()
@@ -257,43 +256,6 @@ private java.util.function.Consumer<Delivery> createNativeHandler(MessageHandler
     };
   }
 
-  private Runnable createReceiveTask(Receiver receiver, MessageHandler messageHandler) {
-    return () -> {
-      try {
-        receiver.addCredit(this.initialCredits);
-        while (!Thread.currentThread().isInterrupted()) {
-          Delivery delivery = receiver.receive(100, TimeUnit.MILLISECONDS);
-          if (delivery != null) {
-            this.unsettledMessageCount.incrementAndGet();
-            this.metricsCollector.consume();
-            AmqpMessage message = new AmqpMessage(delivery.message());
-            Consumer.Context context =
-                new DeliveryContext(
-                    delivery,
-                    this.protonExecutor,
-                    this.metricsCollector,
-                    this.unsettledMessageCount,
-                    this.replenishCreditOperation,
-                    this);
-            messageHandler.handle(context, message);
-          }
-        }
-      } catch (ClientLinkRemotelyClosedException | ClientSessionRemotelyClosedException e) {
-        if (notFound(e) || resourceDeleted(e) || unauthorizedAccess(e)) {
-          this.close(ExceptionUtils.convert(e));
-        }
-      } catch (ClientConnectionRemotelyClosedException e) {
-        // receiver is closed
-      } catch (ClientException e) {
-        java.util.function.Consumer<String> log =
-            this.closed.get() ? m -> LOGGER.debug(m, e) : m -> LOGGER.warn(m, e);
-        log.accept("Error while polling AMQP receiver");
-      } catch (Exception e) {
-        LOGGER.warn("Unexpected error in consumer loop", e);
-      }
-    };
-  }
-
   void recoverAfterConnectionFailure() {
     this.nativeReceiver =
         RetryUtils.callAndMaybeRetry(

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironment.java

@@ -26,9 +26,12 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.qpid.protonj2.client.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 class AmqpEnvironment implements Environment {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(AmqpEnvironment.class);
   private static final AtomicLong ID_SEQUENCE = new AtomicLong(0);
 
   private final Client client;
@@ -128,6 +131,7 @@ CredentialsFactory credentialsFactory() {
   @Override
   public void close() {
     if (this.closed.compareAndSet(false, true)) {
+      LOGGER.debug("Closing environment {}", this);
       this.connectionManager.close();
       this.client.close();
       this.recoveryEventLoop.close();
@@ -145,6 +149,7 @@ public void close() {
         this.clockRefreshFuture.cancel(false);
       }
       this.scheduledExecutorService.shutdownNow();
+      LOGGER.debug("Environment {} has been closed", this);
     }
   }
 

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagement.java

@@ -404,6 +404,7 @@ OutstandingRequest request(Message<?> request, UUID requestId) throws ClientExce
     this.outstandingRequests.put(requestId, outstandingRequest);
     LOGGER.debug("Sending request {}", requestId);
     this.sender.send(request);
+    // FIXME use async callback for management responses
     Future<?> loop = this.receiveLoop;
     if (loop == null) {
       this.instanceLock.lock();

src/main/java/com/rabbitmq/client/amqp/impl/Credentials.java

@@ -28,13 +28,11 @@ interface Registration {
     void connect(AuthenticationCallback callback);
 
     void unregister();
-
   }
 
   interface AuthenticationCallback {
 
     void authenticate(String username, String password);
-
   }
 
   class NoOpCredentials implements Credentials {

src/main/java/com/rabbitmq/client/amqp/impl/CredentialsFactory.java

@@ -92,6 +92,7 @@ private Credentials createOAuthCredentials(DefaultConnectionSettings<?> connecti
             clientBuilderConsumer,
             null,
             new GsonTokenParser());
-    return new TokenCredentials(tokenRequester, environment.scheduledExecutorService());
+    return new TokenCredentials(
+        tokenRequester, environment.scheduledExecutorService(), settings.refreshDelayStrategy());
   }
 }

src/main/java/com/rabbitmq/client/amqp/impl/DefaultConnectionSettings.java

@@ -28,8 +28,10 @@
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManager;
@@ -509,6 +511,8 @@ static class DefaultOAuthSettings<T> implements OAuthSettings<T> {
     private String clientSecret;
     private String grantType = "client_credentials";
     private boolean shared = true;
+    private Function<Instant, Duration> refreshDelayStrategy =
+        TokenCredentials.DEFAULT_REFRESH_DELAY_STRATEGY;
 
     DefaultOAuthSettings(DefaultConnectionSettings<T> connectionSettings) {
       this.connectionSettings = connectionSettings;
@@ -555,6 +559,15 @@ public OAuthSettings<T> shared(boolean shared) {
       return this;
     }
 
+    DefaultOAuthSettings<T> refreshDelayStrategy(Function<Instant, Duration> refreshDelayStrategy) {
+      this.refreshDelayStrategy = refreshDelayStrategy;
+      return this;
+    }
+
+    Function<Instant, Duration> refreshDelayStrategy() {
+      return this.refreshDelayStrategy;
+    }
+
     @Override
     public DefaultOAuthTlsSettings<? extends T> tls() {
       this.tls.enable();
@@ -576,6 +589,7 @@ void copyTo(DefaultOAuthSettings<?> copy) {
       if (this.tls.enabled()) {
         this.tls.copyTo(copy.tls());
       }
+      copy.refreshDelayStrategy(this.refreshDelayStrategy);
     }
 
     String tokenEndpointUri() {