[reactor-optional] Credentials as adapter 8x

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

@@ -6,8 +6,9 @@
  */
 package io.lettuce.authx;
 
+import io.lettuce.core.CredentialsProvider;
 import io.lettuce.core.RedisCredentials;
-import io.lettuce.core.RedisCredentialsProvider;
+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/AsyncCredentialsProviderAdapter.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2026-Present, Redis Ltd. All rights reserved.
+ * SPDX-License-Identifier: MIT
+ */
+package io.lettuce.core;
+
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Adapts a reactor-free {@link CredentialsProvider} to the deprecated reactive {@link RedisCredentialsProvider}, so
+ * {@link RedisURI#getCredentialsProvider()} can keep returning a {@link RedisCredentialsProvider} while credentials are stored
+ * internally as a {@link CredentialsProvider}. Only {@link #resolveCredentials()} materialises a {@link Mono}; the async and
+ * streaming paths delegate directly and stay reactor-free.
+ *
+ * @author Aleksandar Todorov
+ * @since 7.7
+ */
+class AsyncCredentialsProviderAdapter implements RedisCredentialsProvider {
+
+    private final CredentialsProvider delegate;
+
+    AsyncCredentialsProviderAdapter(CredentialsProvider delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public Mono<RedisCredentials> resolveCredentials() {
+        return Mono.fromCompletionStage(delegate.resolveCredentialsAsync());
+    }
+
+    @Override
+    public CompletionStage<RedisCredentials> resolveCredentialsAsync() {
+        return delegate.resolveCredentialsAsync();
+    }
+
+    @Override
+    public boolean supportsStreaming() {
+        return delegate.supportsStreaming();
+    }
+
+    @Override
+    public Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+        return delegate.subscribeToCredentials(onNext, onError);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof AsyncCredentialsProviderAdapter)) {
+            return false;
+        }
+        return delegate.equals(((AsyncCredentialsProviderAdapter) o).delegate);
+    }
+
+    @Override
+    public int hashCode() {
+        return delegate.hashCode();
+    }
+
+}

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;
 
@@ -51,7 +51,7 @@ public class ConnectionState {
     public void apply(RedisURI redisURI) {
 
         connectionMetadata.apply(redisURI);
-        setCredentialsProvider(redisURI.getCredentialsProvider());
+        setCredentialsProvider(redisURI.getCredentialsProviderAsync());
     }
 
     void apply(ConnectionMetadata metadata) {
@@ -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/CredentialsProvider.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) 2026-Present, Redis Ltd. All rights reserved.
+ * SPDX-License-Identifier: MIT
+ */
+package io.lettuce.core;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import io.lettuce.core.internal.Futures;
+import io.lettuce.core.internal.LettuceAssert;
+
+/**
+ * Interface for loading {@link RedisCredentials} used to authenticate a Redis connection, resolving credentials asynchronously
+ * as a {@link CompletionStage}. Replaces {@link RedisCredentialsProvider}.
+ * <p>
+ * Credentials are requested by the driver after connecting to the server. Therefore, credential retrieval is subject to
+ * complete within the connection creation timeout to avoid connection failures.
+ *
+ * @author Aleksandar Todorov
+ * @since 7.7
+ */
+@FunctionalInterface
+public interface CredentialsProvider {
+
+    /**
+     * Returns {@link RedisCredentials} that can be used to authorize a Redis connection. Each implementation of
+     * {@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} that completes with the {@link RedisCredentials} used to authorize a Redis connection.
+     */
+    CompletionStage<RedisCredentials> resolveCredentialsAsync();
+
+    /**
+     * Creates a new {@link CredentialsProvider} from a given {@link Supplier}.
+     *
+     * @param supplier must not be {@code null}.
+     * @return a {@link CredentialsProvider} resolving the {@link RedisCredentials} produced by the {@link Supplier}.
+     */
+    static CredentialsProvider from(Supplier<RedisCredentials> supplier) {
+
+        LettuceAssert.notNull(supplier, "Supplier must not be null");
+
+        return () -> {
+            try {
+                RedisCredentials credentials = supplier.get();
+                if (credentials == null) {
+                    return Futures.failed(new IllegalStateException("Provided RedisCredentials supplier returned null"));
+                }
+                return CompletableFuture.completedFuture(credentials);
+            } catch (Exception e) {
+                return Futures.failed(e);
+            }
+        };
+    }
+
+    /**
+     * 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 CredentialsProvider} supports streaming credentials.
+     */
+    default boolean supportsStreaming() {
+        return false;
+    }
+
+    /**
+     * Subscribe to credential updates produced by this provider.
+     * <p>
+     * For implementations that support streaming credentials (as indicated by {@link #supportsStreaming()} returning
+     * {@code true}), the {@code onNext} consumer is invoked whenever new credentials become available, typically as a result of
+     * external events such as token renewal or rotation. The {@code onError} consumer is invoked when the provider observes a
+     * failure while obtaining new credentials.
+     * <p>
+     * Replay semantics on subscription are defined by the implementation and callers should consult the concrete provider's
+     * documentation. Subscribers must not assume any specific replay behaviour from this interface alone.
+     * <p>
+     * Implementations that do not support streaming credentials (where {@link #supportsStreaming()} returns {@code false})
+     * throw an {@link UnsupportedOperationException} by default.
+     *
+     * @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 Subscription} that can be used to stop receiving updates.
+     * @throws UnsupportedOperationException if the provider does not support streaming credentials.
+     */
+    default Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+        throw new UnsupportedOperationException("Streaming credentials are not supported by this provider.");
+    }
+
+    /**
+     * Extension to {@link CredentialsProvider} that resolves credentials immediately without the need to defer the credential
+     * resolution.
+     */
+    @FunctionalInterface
+    interface ImmediateCredentialsProvider extends CredentialsProvider {
+
+        @Override
+        default CompletionStage<RedisCredentials> resolveCredentialsAsync() {
+            try {
+                RedisCredentials credentials = resolveCredentialsNow();
+                if (credentials == null) {
+                    return Futures.failed(new IllegalStateException("RedisCredentials resolved to null"));
+                }
+                return CompletableFuture.completedFuture(credentials);
+            } catch (Exception e) {
+                return Futures.failed(e);
+            }
+        }
+
+        /**
+         * Returns {@link RedisCredentials} that can be used to authorize a Redis connection. Each implementation of
+         * {@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.
+         */
+        RedisCredentials resolveCredentialsNow();
+
+    }
+
+}