Add OAuth 2 credentials provider and refresh service

WIP.

(cherry picked from commit 4251459)

Conflicts:
	src/test/java/com/rabbitmq/client/test/ClientTests.java

Author: acogoluegnes

pom.xml

@@ -63,6 +63,7 @@
     <awaitility.version>3.1.6</awaitility.version>
     <mockito.version>3.0.0</mockito.version>
     <assertj.version>3.12.2</assertj.version>
+    <jetty.version>9.4.19.v20190610</jetty.version>
 
     <maven.javadoc.plugin.version>3.0.1</maven.javadoc.plugin.version>
     <maven.release.plugin.version>2.5.3</maven.release.plugin.version>
@@ -744,6 +745,12 @@
       <version>1.3</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-servlet</artifactId>
+      <version>${jetty.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -15,15 +15,7 @@
 
 package com.rabbitmq.client;
 
-import com.rabbitmq.client.impl.AMQConnection;
-import com.rabbitmq.client.impl.ConnectionParams;
-import com.rabbitmq.client.impl.CredentialsProvider;
-import com.rabbitmq.client.impl.DefaultCredentialsProvider;
-import com.rabbitmq.client.impl.DefaultExceptionHandler;
-import com.rabbitmq.client.impl.ErrorOnWriteListener;
-import com.rabbitmq.client.impl.FrameHandler;
-import com.rabbitmq.client.impl.FrameHandlerFactory;
-import com.rabbitmq.client.impl.SocketFrameHandlerFactory;
+import com.rabbitmq.client.impl.*;
 import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.client.impl.nio.SocketChannelFrameHandlerFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
@@ -209,6 +201,8 @@ public class ConnectionFactory implements Cloneable {
      */
     private TrafficListener trafficListener = TrafficListener.NO_OP;
 
+    private CredentialsRefreshService credentialsRefreshService;
+
     /** @return the default host to use for connections */
     public String getHost() {
         return host;
@@ -854,6 +848,10 @@ public MetricsCollector getMetricsCollector() {
         return metricsCollector;
     }
 
+    public void setCredentialsRefreshService(CredentialsRefreshService credentialsRefreshService) {
+        this.credentialsRefreshService = credentialsRefreshService;
+    }
+
     protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IOException {
         if(nio) {
             if(this.frameHandlerFactory == null) {
@@ -1161,6 +1159,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setConnectionRecoveryTriggeringCondition(connectionRecoveryTriggeringCondition);
         result.setTopologyRecoveryRetryHandler(topologyRecoveryRetryHandler);
         result.setTrafficListener(trafficListener);
+        result.setCredentialsRefreshService(credentialsRefreshService);
         return result;
     }
 

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -142,6 +142,7 @@ public static Map<String, Object> defaultClientProperties() {
     private final int channelRpcTimeout;
     private final boolean channelShouldCheckRpcResponseType;
     private final TrafficListener trafficListener;
+    private final CredentialsRefreshService credentialsRefreshService;
 
     /* State modified after start - all volatile */
 
@@ -239,6 +240,9 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
         this.channelShouldCheckRpcResponseType = params.channelShouldCheckRpcResponseType();
 
         this.trafficListener = params.getTrafficListener() == null ? TrafficListener.NO_OP : params.getTrafficListener();
+
+        this.credentialsRefreshService = params.getCredentialsRefreshService();
+
         this._channel0 = new AMQChannel(this, 0) {
             @Override public boolean processAsync(Command c) throws IOException {
                 return getConnection().processControlCommand(c);
@@ -336,6 +340,15 @@ public void start()
 
             String username = credentialsProvider.getUsername();
             String password = credentialsProvider.getPassword();
+
+            if (credentialsProvider.getExpiration() != null) {
+                if (this.credentialsRefreshService.needRefresh(credentialsProvider.getExpiration())) {
+                    credentialsProvider.refresh();
+                    username = credentialsProvider.getUsername();
+                    password = credentialsProvider.getPassword();
+                }
+            }
+
             LongString challenge = null;
             LongString response = sm.handleChallenge(null, username, password);
 
@@ -413,6 +426,26 @@ public void start()
             throw AMQChannel.wrap(sse);
         }
 
+        if (this.credentialsProvider.getExpiration() != null) {
+            String registrationId = this.credentialsRefreshService.register(credentialsProvider, () -> {
+                // return false if connection is closed, so refresh service can get rid of this registration
+                if (!isOpen()) {
+                    return false;
+                }
+                if (this._inConnectionNegotiation) {
+                    // this should not happen
+                    return true;
+                }
+                String refreshedPassword = credentialsProvider.getPassword();
+
+                // TODO send password to server with update-secret extension, using channel 0
+
+                return true;
+            });
+
+            addShutdownListener(sse -> this.credentialsRefreshService.unregister(this.credentialsProvider, registrationId));
+        }
+
         // We can now respond to errors having finished tailoring the connection
         this._inConnectionNegotiation = false;
     }

src/main/java/com/rabbitmq/client/impl/ConnectionParams.java

@@ -60,6 +60,8 @@ public class ConnectionParams {
 
     private TrafficListener trafficListener;
 
+    private CredentialsRefreshService credentialsRefreshService;
+
     public ConnectionParams() {}
 
     public CredentialsProvider getCredentialsProvider() {
@@ -277,4 +279,12 @@ public void setTrafficListener(TrafficListener trafficListener) {
     public TrafficListener getTrafficListener() {
         return trafficListener;
     }
+
+    public void setCredentialsRefreshService(CredentialsRefreshService credentialsRefreshService) {
+        this.credentialsRefreshService = credentialsRefreshService;
+    }
+
+    public CredentialsRefreshService getCredentialsRefreshService() {
+        return credentialsRefreshService;
+    }
 }

src/main/java/com/rabbitmq/client/impl/CredentialsProvider.java

@@ -1,16 +1,67 @@
+// Copyright (c) 2018-2019 Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
 package com.rabbitmq.client.impl;
 
+import java.util.Date;
+
 /**
  * Provider interface for establishing credentials for connecting to the broker. Especially useful
- * for situations where credentials might change before a recovery takes place or where it is 
+ * for situations where credentials might expire or change before a recovery takes place or where it is
  * convenient to plug in an outside custom implementation.
  *
- * @since 4.5.0
+ * @see CredentialsRefreshService
+ * @since 5.2.0
  */
 public interface CredentialsProvider {
 
+    /**
+     * Username to use for authentication
+     *
+     * @return username
+     */
     String getUsername();
 
+    /**
+     * Password/secret/token to use for authentication
+     *
+     * @return password/secret/token
+     */
     String getPassword();
 
+    /**
+     * The expiration date of the credentials, if any.
+     * <p>
+     * If credentials do not expire, must return null. Default
+     * behavior is to return null, assuming credentials never
+     * expire.
+     *
+     * @return credentials expiration date
+     */
+    default Date getExpiration() {
+        // no expiration by default
+        return null;
+    }
+
+    /**
+     * Instructs the provider to refresh or renew credentials.
+     * <p>
+     * Default behavior is no-op.
+     */
+    default void refresh() {
+        // no need to refresh anything by default
+    }
+
 }

src/main/java/com/rabbitmq/client/impl/CredentialsRefreshService.java

@@ -0,0 +1,73 @@
+// Copyright (c) 2019 Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
+package com.rabbitmq.client.impl;
+
+import java.util.Date;
+import java.util.concurrent.Callable;
+
+/**
+ * Provider interface to refresh credentials when appropriate
+ * and perform an operation once the credentials have been
+ * renewed. In the context of RabbitMQ, the operation consists
+ * in calling the <code>update.secret</code> AMQP extension
+ * to provide new valid credentials before the current ones
+ * expire.
+ * <p>
+ * New connections are registered and implementations must perform
+ * credentials renewal when appropriate. Implementations
+ * must call a registered callback once credentials are renewed.
+ *
+ * @see CredentialsProvider
+ * @see DefaultCredentialsRefreshService
+ */
+public interface CredentialsRefreshService {
+
+    /**
+     * Register a new entity that needs credentials renewal.
+     * <p>
+     * The registered callback must return true if the action was
+     * performed correctly, throw an exception if something goes wrong,
+     * and return false if it became stale and wants to be unregistered.
+     * <p>
+     * Implementations are free to automatically unregister an entity whose
+     * callback has failed a given number of times.
+     *
+     * @param credentialsProvider the credentials provider
+     * @param refreshAction       the action to perform after credentials renewal
+     * @return a tracking ID for the registration
+     */
+    String register(CredentialsProvider credentialsProvider, Callable<Boolean> refreshAction);
+
+    /**
+     * Unregister the entity with the given registration ID.
+     * <p>
+     * Its state is cleaned up and its registered callback will not be
+     * called again.
+     *
+     * @param credentialsProvider the credentials provider
+     * @param registrationId      the registration ID
+     */
+    void unregister(CredentialsProvider credentialsProvider, String registrationId);
+
+    /**
+     * Provide a hint about whether credentials should be renewed.
+     *
+     * @param expiration
+     * @return true if credentials should be renewed, false otherwise
+     */
+    boolean needRefresh(Date expiration);
+
+}

src/main/java/com/rabbitmq/client/impl/DefaultCredentialsProvider.java

@@ -1,3 +1,18 @@
+// Copyright (c) 2018-2019 Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
 package com.rabbitmq.client.impl;
 
 /**