Start support for OAuth 2 token retrieval and refresh

Author: acogoluegnes

pom.xml

@@ -44,7 +44,6 @@
         <slf4j.version>1.7.36</slf4j.version>
         <netty4.version>4.1.115.Final</netty4.version>
         <netty4.iouring.version>0.0.25.Final</netty4.iouring.version>
-        <netty5.version>5.0.0.Alpha5</netty5.version>
         <micrometer.version>1.14.1</micrometer.version>
         <logback.version>1.2.13</logback.version>
         <junit.jupiter.version>5.11.3</junit.jupiter.version>
@@ -77,6 +76,7 @@
         <gpg.skip>true</gpg.skip>
         <gpg.keyname>6026DFCA</gpg.keyname>
         <spotbugs.skip>false</spotbugs.skip>
+        <gson.version>2.11.0</gson.version>
     </properties>
 
     <dependencies>
@@ -140,6 +140,13 @@
         </dependency>
         <!-- End of QPid dependencies -->
 
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>${gson.version}</version>
+            <optional>true</optional>
+        </dependency>
+
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-engine</artifactId>

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

@@ -76,6 +76,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final Lock instanceLock = new ReentrantLock();
   private final boolean filterExpressionsSupported, setTokenSupported;
   private volatile ExecutorService dispatchingExecutorService;
+  private final Credentials credentials;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
     super(builder.listeners());
@@ -114,6 +115,14 @@ final class AmqpConnection extends ResourceBase implements Connection {
       this.affinityStrategy = null;
     }
     this.management = createManagement();
+    if (this.connectionSettings.credentialsProvider()
+        instanceof UsernamePasswordCredentialsProvider) {
+      UsernamePasswordCredentialsProvider credentialsProvider =
+          (UsernamePasswordCredentialsProvider) connectionSettings.credentialsProvider();
+      this.credentials = new UsernamePasswordCredentials(credentialsProvider);
+    } else {
+      this.credentials = callback -> {};
+    }
     LOGGER.debug("Opening native connection for connection '{}'...", this.name());
     NativeConnectionWrapper ncw =
         ConnectionUtils.enforceAffinity(
@@ -190,12 +199,7 @@ private NativeConnectionWrapper connect(
       List<Address> addresses) {
 
     ConnectionOptions connectionOptions = new ConnectionOptions();
-    if (connectionSettings.credentialsProvider() instanceof UsernamePasswordCredentialsProvider) {
-      UsernamePasswordCredentialsProvider credentialsProvider =
-          (UsernamePasswordCredentialsProvider) connectionSettings.credentialsProvider();
-      connectionOptions.user(credentialsProvider.getUsername());
-      connectionOptions.password(credentialsProvider.getPassword());
-    }
+    credentials.configure(new TokenConnectionCallback(connectionOptions));
     connectionOptions.virtualHost("vhost:" + connectionSettings.virtualHost());
     connectionOptions.saslOptions().addAllowedMechanism(connectionSettings.saslMechanism());
     connectionOptions.idleTimeout(
@@ -807,4 +811,25 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hashCode(id);
   }
+
+  private static class TokenConnectionCallback implements Credentials.ConnectionCallback {
+
+    private final ConnectionOptions options;
+
+    private TokenConnectionCallback(ConnectionOptions options) {
+      this.options = options;
+    }
+
+    @Override
+    public Credentials.ConnectionCallback username(String username) {
+      options.user(username);
+      return this;
+    }
+
+    @Override
+    public Credentials.ConnectionCallback password(String password) {
+      options.password(password);
+      return this;
+    }
+  }
 }

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

@@ -333,7 +333,9 @@ void close(Throwable cause) {
       this.state(CLOSING, cause);
       this.connection.removeConsumer(this);
       try {
-        this.nativeReceiver.close();
+        if (this.nativeReceiver != null) {
+          this.nativeReceiver.close();
+        }
         this.sessionHandler.close();
       } catch (Exception e) {
         LOGGER.warn("Error while closing receiver", e);

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

@@ -210,7 +210,9 @@ void close(Throwable cause) {
       this.state(State.CLOSING, cause);
       this.connection.removePublisher(this);
       try {
-        this.sender.close();
+        if (this.sender != null) {
+          this.sender.close();
+        }
         this.sessionHandler.close();
       } catch (Exception e) {
         LOGGER.warn("Error while closing sender", e);

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

@@ -0,0 +1,30 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.impl;
+
+interface Credentials {
+
+  void configure(ConnectionCallback callback);
+
+  interface ConnectionCallback {
+
+    ConnectionCallback username(String username);
+
+    ConnectionCallback password(String password);
+  }
+}

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

@@ -0,0 +1,56 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.impl;
+
+import com.rabbitmq.client.amqp.UsernamePasswordCredentialsProvider;
+import com.rabbitmq.client.amqp.oauth.Token;
+import com.rabbitmq.client.amqp.oauth.TokenParser;
+import com.rabbitmq.client.amqp.oauth.TokenRequester;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+final class OAuthCredentialsProvider implements UsernamePasswordCredentialsProvider {
+
+  private final TokenRequester requester;
+  private final TokenParser parser;
+  private volatile Token token;
+  private final Lock lock = new ReentrantLock();
+
+  OAuthCredentialsProvider(TokenRequester requester, TokenParser parser) {
+    this.requester = requester;
+    this.parser = parser;
+  }
+
+  @Override
+  public String getUsername() {
+    return "";
+  }
+
+  @Override
+  public String getPassword() {
+    lock.lock();
+    try {
+      if (token == null || token.expirationTime() < System.currentTimeMillis()) {
+        this.token = this.parser.parse(this.requester.request());
+      }
+    } finally {
+      lock.unlock();
+    }
+    return token.value();
+  }
+}

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

@@ -0,0 +1,34 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.impl;
+
+import com.rabbitmq.client.amqp.UsernamePasswordCredentialsProvider;
+
+final class UsernamePasswordCredentials implements Credentials {
+
+  private final UsernamePasswordCredentialsProvider provider;
+
+  UsernamePasswordCredentials(UsernamePasswordCredentialsProvider provider) {
+    this.provider = provider;
+  }
+
+  @Override
+  public void configure(ConnectionCallback callback) {
+    callback.username(this.provider.getUsername()).password(this.provider.getPassword());
+  }
+}

src/main/java/com/rabbitmq/client/amqp/oauth/GsonTokenParser.java

@@ -0,0 +1,59 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.oauth;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import java.util.Map;
+
+public class GsonTokenParser implements TokenParser {
+
+  private static final Gson GSON = new Gson();
+  private static final TypeToken<Map<String, Object>> MAP_TYPE = new TypeToken<>() {};
+
+  @Override
+  public Token parse(String tokenAsString) {
+    Map<String, Object> tokenAsMap = GSON.fromJson(tokenAsString, MAP_TYPE);
+    String accessToken = (String) tokenAsMap.get("access_token");
+    // in seconds, see https://www.rfc-editor.org/rfc/rfc6749#section-5.1
+    long expiresIn = ((Number) tokenAsMap.get("expires_in")).longValue();
+    long expirationTime = System.currentTimeMillis() + expiresIn * 1_000;
+    return new DefaultTokenInfo(accessToken, expirationTime);
+  }
+
+  private static final class DefaultTokenInfo implements Token {
+
+    private final String value;
+    private final long expirationTime;
+
+    private DefaultTokenInfo(String value, long expirationTime) {
+      this.value = value;
+      this.expirationTime = expirationTime;
+    }
+
+    @Override
+    public String value() {
+      return this.value;
+    }
+
+    @Override
+    public long expirationTime() {
+      return this.expirationTime;
+    }
+  }
+}