[reactor-optional] Remove RedisCredentialsProvider and promote CredentialsProvider as primary

src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java

@@ -7,7 +7,8 @@
 package io.lettuce.authx;
 
 import io.lettuce.core.RedisCredentials;
-import io.lettuce.core.RedisCredentialsProvider;
+import io.lettuce.core.CredentialsProvider;
+import io.lettuce.core.Subscription;
 import io.lettuce.core.internal.LettuceAssert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,7 +26,7 @@
 import java.util.function.Consumer;
 
 /**
- * A {@link RedisCredentialsProvider} implementation that supports token-based authentication for Redis.
+ * A {@link CredentialsProvider} implementation that supports token-based authentication for Redis.
  * <p>
  * This provider uses a {@link TokenManager} to manage and renew tokens, ensuring that the Redis client can authenticate with
  * Redis using a dynamically updated token. This is particularly useful in scenarios where Redis access is controlled via
@@ -48,11 +49,11 @@
  *
  * @since 6.6
  */
-public class TokenBasedRedisCredentialsProvider implements RedisCredentialsProvider, AutoCloseable {
+public class TokenBasedRedisCredentialsProvider implements CredentialsProvider, AutoCloseable {
 
     private static final Logger log = LoggerFactory.getLogger(TokenBasedRedisCredentialsProvider.class);
 
-    private static class SimpleSubscription implements CredentialsSubscription {
+    private static class SimpleSubscription implements Subscription {
 
         private final TokenBasedRedisCredentialsProvider provider;
 
@@ -205,7 +206,7 @@ private static void dispatchOnError(SimpleSubscription subscription, Throwable t
      * @return a {@link CompletionStage} that completes with the latest Redis credentials
      */
     @Override
-    public CompletionStage<RedisCredentials> resolveCredentials() {
+    public CompletionStage<RedisCredentials> resolveCredentialsAsync() {
         CompletableFuture<RedisCredentials> result = new CompletableFuture<>();
         credentialsFutureRef.get().whenComplete((creds, throwable) -> {
             if (throwable != null) {
@@ -218,7 +219,7 @@ public CompletionStage<RedisCredentials> resolveCredentials() {
     }
 
     @Override
-    public CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+    public Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
         if (isClosed) {
             throw new IllegalStateException("Credentials provider closed");
         }
@@ -280,7 +281,7 @@ public void close() {
      * <ul>
      * <li>Subscriber {@code onNext}/{@code onError} callbacks for live token renewals run on the {@link TokenManager}'s renewal
      * thread. A slow or blocking subscriber can delay or miss subsequent renewals.</li>
-     * <li>Continuations chained off {@link #resolveCredentials()} run on the renewal thread when the initial future
+     * <li>Continuations chained off {@link #resolveCredentialsAsync()} run on the renewal thread when the initial future
      * completes.</li>
      * <li>Replay deliveries to a newly subscribing consumer run on the subscribing thread.</li>
      * </ul>
@@ -323,7 +324,7 @@ public static TokenBasedRedisCredentialsProvider create(TokenAuthConfig tokenAut
      * <ul>
      * <li>Subscriber {@code onNext}/{@code onError} callbacks for live token renewals run on the {@link TokenManager}'s renewal
      * thread. A slow or blocking subscriber can delay or miss subsequent renewals.</li>
-     * <li>Continuations chained off {@link #resolveCredentials()} run on the renewal thread when the initial future
+     * <li>Continuations chained off {@link #resolveCredentialsAsync()} run on the renewal thread when the initial future
      * completes.</li>
      * <li>Replay deliveries to a newly subscribing consumer run on the subscribing thread.</li>
      * </ul>

src/main/java/io/lettuce/core/ClientOptions.java

@@ -600,7 +600,7 @@ public DisconnectedBehavior getDisconnectedBehavior() {
     }
 
     /**
-     * Behavior for re-authentication when the {@link RedisCredentialsProvider} emits new credentials. Defaults to
+     * Behavior for re-authentication when the {@link CredentialsProvider} emits new credentials. Defaults to
      * {@link ReauthenticateBehavior#DEFAULT}.
      *
      * @return the currently set {@link ReauthenticateBehavior}.
@@ -743,31 +743,31 @@ public TimeoutOptions getTimeoutOptions() {
     /**
      * Defines the re-authentication behavior of the Redis client.
      * <p/>
-     * Certain implementations of the {@link RedisCredentialsProvider} could emit new credentials at runtime. This setting
-     * controls how the driver reacts to these newly emitted credentials.
+     * Certain implementations of the {@link CredentialsProvider} could emit new credentials at runtime. This setting controls
+     * how the driver reacts to these newly emitted credentials.
      */
     public enum ReauthenticateBehavior {
 
         /**
          * This is the default behavior. The client will fetch current credentials from the underlying
-         * {@link RedisCredentialsProvider} only when the driver needs to, e.g. when the connection is first established or when
-         * it is re-established after a disconnect.
+         * {@link CredentialsProvider} only when the driver needs to, e.g. when the connection is first established or when it
+         * is re-established after a disconnect.
          * <p/>
          * <p>
-         * No re-authentication is performed when new credentials are emitted by a {@link RedisCredentialsProvider} that
-         * supports streaming. The client does not subscribe to or react to any updates published by
-         * {@link RedisCredentialsProvider#subscribeToCredentials(java.util.function.Consumer, java.util.function.Consumer)}.
+         * No re-authentication is performed when new credentials are emitted by a {@link CredentialsProvider} that supports
+         * streaming. The client does not subscribe to or react to any updates published by
+         * {@link CredentialsProvider#subscribeToCredentials(java.util.function.Consumer, java.util.function.Consumer)}.
          * </p>
          */
         DEFAULT,
 
         /**
-         * Automatically triggers re-authentication whenever new credentials are emitted by a {@link RedisCredentialsProvider}
-         * that supports streaming, as indicated by {@link RedisCredentialsProvider#supportsStreaming()}.
+         * Automatically triggers re-authentication whenever new credentials are emitted by a {@link CredentialsProvider} that
+         * supports streaming, as indicated by {@link CredentialsProvider#supportsStreaming()}.
          *
          * <p>
          * When this behavior is enabled, the client subscribes to credential updates via
-         * {@link RedisCredentialsProvider#subscribeToCredentials(java.util.function.Consumer, java.util.function.Consumer)} and
+         * {@link CredentialsProvider#subscribeToCredentials(java.util.function.Consumer, java.util.function.Consumer)} and
          * issues an {@code AUTH} command to the Redis server each time new credentials are received. This behavior supports
          * dynamic credential scenarios, such as token-based authentication, or credential rotation where credentials are
          * refreshed periodically to maintain access.

src/main/java/io/lettuce/core/ConnectionState.java

@@ -35,7 +35,7 @@ public class ConnectionState {
 
     private volatile HandshakeResponse handshakeResponse;
 
-    private volatile RedisCredentialsProvider credentialsProvider;
+    private volatile CredentialsProvider credentialsProvider;
 
     private volatile int db;
 
@@ -129,11 +129,11 @@ protected void setUserNamePassword(List<char[]> args) {
         }
     }
 
-    protected void setCredentialsProvider(RedisCredentialsProvider credentialsProvider) {
+    protected void setCredentialsProvider(CredentialsProvider credentialsProvider) {
         this.credentialsProvider = credentialsProvider;
     }
 
-    public RedisCredentialsProvider getCredentialsProvider() {
+    public CredentialsProvider getCredentialsProvider() {
         return credentialsProvider;
     }
 

src/main/java/io/lettuce/core/RedisCredentialsProvider.java → src/main/java/io/lettuce/core/CredentialsProvider.java

@@ -1,6 +1,5 @@
 package io.lettuce.core;
 
-import java.io.Closeable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
@@ -20,36 +19,27 @@
  * @since 6.2
  */
 @FunctionalInterface
-public interface RedisCredentialsProvider {
-
-    /**
-     * Handle to a subscription created by {@link #subscribeToCredentials(Consumer, Consumer)}. Closing the subscription stops
-     * the provider from delivering further credential updates to the registered consumers.
-     */
-    interface CredentialsSubscription extends Closeable {
-
-        @Override
-        void close();
-
-    }
+public interface CredentialsProvider {
 
     /**
      * Returns {@link RedisCredentials} that can be used to authorize a Redis connection. Each implementation of
-     * {@code RedisCredentialsProvider} can choose its own strategy for loading credentials. For example, an implementation
-     * might load credentials from an existing key management system, or load new credentials when credentials are rotated. If
-     * an error occurs during the loading of credentials or credentials could not be found, a runtime exception will be raised.
+     * {@code CredentialsProvider} can choose its own strategy for loading credentials. For example, an implementation might
+     * load credentials from an existing key management system, or load new credentials when credentials are rotated. If an
+     * error occurs during the loading of credentials or credentials could not be found, the returned {@link CompletionStage}
+     * completes exceptionally.
      *
-     * @return a {@link CompletionStage} emitting {@link RedisCredentials} that can be used to authorize a Redis connection.
+     * @return a {@link CompletionStage} that completes with the {@link RedisCredentials} used to authorize a Redis connection.
+     * @since 7.7
      */
-    CompletionStage<RedisCredentials> resolveCredentials();
+    CompletionStage<RedisCredentials> resolveCredentialsAsync();
 
     /**
-     * Creates a new {@link RedisCredentialsProvider} from a given {@link Supplier}.
+     * Creates a new {@link CredentialsProvider} from a given {@link Supplier}.
      *
      * @param supplier must not be {@code null}.
      * @return the {@link RedisCredentials} using credentials from {@link Supplier}.
      */
-    static RedisCredentialsProvider from(Supplier<RedisCredentials> supplier) {
+    static CredentialsProvider from(Supplier<RedisCredentials> supplier) {
 
         LettuceAssert.notNull(supplier, "Supplier must not be null");
 
@@ -67,11 +57,11 @@ static RedisCredentialsProvider from(Supplier<RedisCredentials> supplier) {
     }
 
     /**
-     * Some implementations of the {@link RedisCredentialsProvider} may support streaming new credentials, based on some event
-     * that originates outside the driver. In this case they should indicate that so the {@link RedisAuthenticationHandler} is
-     * able to process these new credentials.
+     * Some implementations of the {@link CredentialsProvider} may support streaming new credentials, based on some event that
+     * originates outside the driver. In this case they should indicate that so the {@link RedisAuthenticationHandler} is able
+     * to process these new credentials.
      * 
-     * @return whether the {@link RedisCredentialsProvider} supports streaming credentials.
+     * @return whether the {@link CredentialsProvider} supports streaming credentials.
      */
     default boolean supportsStreaming() {
         return false;
@@ -101,22 +91,22 @@ default boolean supportsStreaming() {
      *
      * @param onNext consumer invoked with each new {@link RedisCredentials} value, must not be {@code null}.
      * @param onError consumer invoked with errors observed while producing credentials, must not be {@code null}.
-     * @return a {@link CredentialsSubscription} that can be used to stop receiving updates.
+     * @return a {@link Subscription} that can be used to stop receiving updates.
      * @throws UnsupportedOperationException if the provider does not support streaming credentials.
      */
-    default CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+    default Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
         throw new UnsupportedOperationException("Streaming credentials are not supported by this provider.");
     }
 
     /**
-     * Extension to {@link RedisCredentialsProvider} that resolves credentials immediately without the need to defer the
-     * credential resolution.
+     * Extension to {@link CredentialsProvider} that resolves credentials immediately without the need to defer the credential
+     * resolution.
      */
     @FunctionalInterface
-    interface ImmediateRedisCredentialsProvider extends RedisCredentialsProvider {
+    interface ImmediateRedisCredentialsProvider extends CredentialsProvider {
 
         @Override
-        default CompletionStage<RedisCredentials> resolveCredentials() {
+        default CompletionStage<RedisCredentials> resolveCredentialsAsync() {
             try {
                 RedisCredentials credentials = resolveCredentialsNow();
                 if (credentials == null) {
@@ -130,10 +120,9 @@ default CompletionStage<RedisCredentials> resolveCredentials() {
 
         /**
          * Returns {@link RedisCredentials} that can be used to authorize a Redis connection. Each implementation of
-         * {@code RedisCredentialsProvider} can choose its own strategy for loading credentials. For example, an implementation
-         * might load credentials from an existing key management system, or load new credentials when credentials are rotated.
-         * If an error occurs during the loading of credentials or credentials could not be found, a runtime exception will be
-         * raised.
+         * {@code CredentialsProvider} can choose its own strategy for loading credentials. For example, an implementation might
+         * load credentials from an existing key management system, or load new credentials when credentials are rotated. If an
+         * error occurs during the loading of credentials or credentials could not be found, a runtime exception will be raised.
          *
          * @return the resolved {@link RedisCredentials} that can be used to authorize a Redis connection.
          */

src/main/java/io/lettuce/core/RedisAuthenticationHandler.java

@@ -6,7 +6,6 @@
  */
 package io.lettuce.core;
 
-import io.lettuce.core.RedisCredentialsProvider.CredentialsSubscription;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.event.connection.ReauthenticationEvent;
@@ -46,9 +45,9 @@ public class RedisAuthenticationHandler<K, V> {
 
     private final StatefulRedisConnectionImpl<K, V> connection;
 
-    private final RedisCredentialsProvider credentialsProvider;
+    private final CredentialsProvider credentialsProvider;
 
-    private final AtomicReference<CredentialsSubscription> credentialsSubscription = new AtomicReference<>();
+    private final AtomicReference<Subscription> credentialsSubscription = new AtomicReference<>();
 
     private final Boolean isPubSubConnection;
 
@@ -62,11 +61,11 @@ public class RedisAuthenticationHandler<K, V> {
      * Creates a new {@link RedisAuthenticationHandler}.
      *
      * @param connection the connection to authenticate
-     * @param credentialsProvider the implementation of {@link RedisCredentialsProvider} to use
+     * @param credentialsProvider the implementation of {@link CredentialsProvider} to use
      * @param isPubSubConnection {@code true} if the connection is a pub/sub connection
      */
-    public RedisAuthenticationHandler(StatefulRedisConnectionImpl<K, V> connection,
-            RedisCredentialsProvider credentialsProvider, Boolean isPubSubConnection) {
+    public RedisAuthenticationHandler(StatefulRedisConnectionImpl<K, V> connection, CredentialsProvider credentialsProvider,
+            Boolean isPubSubConnection) {
         this.connection = connection;
         this.credentialsProvider = credentialsProvider;
         this.isPubSubConnection = isPubSubConnection;
@@ -76,16 +75,16 @@ public RedisAuthenticationHandler(StatefulRedisConnectionImpl<K, V> connection,
      * Creates a new {@link RedisAuthenticationHandler} if the connection supports re-authentication.
      *
      * @param connection the connection to authenticate
-     * @param credentialsProvider the implementation of {@link RedisCredentialsProvider} to use
+     * @param credentialsProvider the implementation of {@link CredentialsProvider} to use
      * @param isPubSubConnection {@code true} if the connection is a pub/sub connection
      * @param options the {@link ClientOptions} to use
      * @return a new {@link RedisAuthenticationHandler} if the connection supports re-authentication, otherwise an
      *         implementation of the {@link RedisAuthenticationHandler} that does nothing
      * @since 6.6.0
-     * @see RedisCredentialsProvider
+     * @see CredentialsProvider
      */
     public static <K, V> RedisAuthenticationHandler<K, V> createHandler(StatefulRedisConnectionImpl<K, V> connection,
-            RedisCredentialsProvider credentialsProvider, Boolean isPubSubConnection, ClientOptions options) {
+            CredentialsProvider credentialsProvider, Boolean isPubSubConnection, ClientOptions options) {
 
         if (isSupported(options)) {
 
@@ -107,7 +106,7 @@ public static <K, V> RedisAuthenticationHandler<K, V> createHandler(StatefulRedi
      *
      * @return a new {@link RedisAuthenticationHandler}
      * @since 6.6.0
-     * @see RedisCredentialsProvider
+     * @see CredentialsProvider
      */
     public static <K, V> RedisAuthenticationHandler<K, V> createDefaultAuthenticationHandler() {
         return new DisabledAuthenticationHandler<>();
@@ -128,10 +127,9 @@ public void subscribe() {
             return;
         }
 
-        CredentialsSubscription credentialsSub = credentialsProvider.subscribeToCredentials(this::reauthenticate,
-                this::onError);
+        Subscription credentialsSub = credentialsProvider.subscribeToCredentials(this::reauthenticate, this::onError);
 
-        CredentialsSubscription oldSub = credentialsSubscription.getAndSet(credentialsSub);
+        Subscription oldSub = credentialsSubscription.getAndSet(credentialsSub);
         if (oldSub != null) {
             try {
                 oldSub.close();
@@ -145,7 +143,7 @@ public void subscribe() {
      * Unsubscribes from the current credentials stream.
      */
     public void unsubscribe() {
-        CredentialsSubscription sub = credentialsSubscription.getAndSet(null);
+        Subscription sub = credentialsSubscription.getAndSet(null);
         if (sub != null) {
             try {
                 sub.close();
@@ -165,7 +163,7 @@ protected void reauthenticate(RedisCredentials credentials) {
     }
 
     /**
-     * Handles errors observed by the underlying {@link RedisCredentialsProvider} while producing credentials.
+     * Handles errors observed by the underlying {@link CredentialsProvider} while producing credentials.
      *
      * @param e the error reported by the credentials provider
      */
@@ -361,7 +359,7 @@ private String getEpid() {
     private static final class DisabledAuthenticationHandler<K, V> extends RedisAuthenticationHandler<K, V> {
 
         public DisabledAuthenticationHandler(StatefulRedisConnectionImpl<K, V> connection,
-                RedisCredentialsProvider credentialsProvider, Boolean isPubSubConnection) {
+                CredentialsProvider credentialsProvider, Boolean isPubSubConnection) {
             super(null, null, null);
         }