diff --git a/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java b/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java index 81b0bc9dd5..5ef7878e19 100644 --- a/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java +++ b/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java @@ -120,6 +120,7 @@ public static RedisAuthenticationHandler createDefaultAuthenticatio * Each time new credentials are received, the client is re-authenticated. The previous subscription, if any, is disposed of * before setting the new subscription. */ + @SuppressWarnings("deprecation") public void subscribe() { if (credentialsProvider == null || !credentialsProvider.supportsStreaming()) { return; diff --git a/src/main/java/io/lettuce/core/RedisCredentialsProvider.java b/src/main/java/io/lettuce/core/RedisCredentialsProvider.java index 9c57a280af..4f5f0473fb 100644 --- a/src/main/java/io/lettuce/core/RedisCredentialsProvider.java +++ b/src/main/java/io/lettuce/core/RedisCredentialsProvider.java @@ -1,10 +1,13 @@ package io.lettuce.core; +import java.util.concurrent.CompletionStage; +import java.util.function.Consumer; import java.util.function.Supplier; +import io.lettuce.core.internal.LettuceAssert; +import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import io.lettuce.core.internal.LettuceAssert; /** * Interface for loading {@link RedisCredentials} that are used for authentication. A commonly-used implementation is @@ -26,9 +29,26 @@ public interface RedisCredentialsProvider { * an error occurs during the loading of credentials or credentials could not be found, a runtime exception will be raised. * * @return a {@link Mono} emitting {@link RedisCredentials} that can be used to authorize a Redis connection. + * @deprecated since 7.7, use {@link #resolveCredentialsAsync()} instead. This Reactor-typed method is removed in Lettuce + * 8.0, when {@code reactor-core} becomes optional and {@link #resolveCredentialsAsync()} becomes the primary + * credential-resolution contract. */ + @Deprecated Mono resolveCredentials(); + /** + * Resolve the latest available credentials as a {@link CompletionStage}. This Reactor-free method replaces + * {@link #resolveCredentials()} and becomes the primary credential-resolution contract in Lettuce 8.0. Prefer it for new + * code; implementations continue to supply credentials through {@link #resolveCredentials()} until 8.0. + * + * @return a {@link CompletionStage} that completes with the {@link RedisCredentials} used to authorize a Redis connection. + * @since 7.7 + */ + @SuppressWarnings("deprecation") + default CompletionStage resolveCredentialsAsync() { + return resolveCredentials().toFuture(); + } + /** * Creates a new {@link RedisCredentialsProvider} from a given {@link Supplier}. * @@ -46,7 +66,7 @@ static RedisCredentialsProvider from(Supplier 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. - * + * * @return whether the {@link RedisCredentialsProvider} supports streaming credentials. */ default boolean supportsStreaming() { @@ -65,11 +85,35 @@ default boolean supportsStreaming() { * * @return a {@link Flux} emitting {@link RedisCredentials}, or throws an exception if streaming is not supported. * @throws UnsupportedOperationException if the provider does not support streaming credentials. + * @deprecated since 7.7, use {@link #subscribeToCredentials(Consumer, Consumer)} instead; scheduled for removal in Lettuce + * 8.0 (when {@code reactor-core} becomes optional). */ + @Deprecated default Flux credentials() { throw new UnsupportedOperationException("Streaming credentials are not supported by this provider."); } + /** + * Subscribe to credential updates produced by this provider. For providers that support streaming (as indicated by + * {@link #supportsStreaming()} returning {@code true}), {@code onNext} is invoked whenever new credentials become available + * (e.g. token renewal or rotation) and {@code onError} is invoked when the provider observes a failure while producing + * credentials. Delivery stops once the returned {@link Subscription} is {@link Subscription#close() closed}. + *

+ * Providers that do not support streaming 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 stops delivery when closed. + * @throws UnsupportedOperationException if the provider does not support streaming credentials. + * @since 7.7 + */ + default Subscription subscribeToCredentials(Consumer onNext, Consumer onError) { + LettuceAssert.notNull(onNext, "onNext consumer must not be null"); + LettuceAssert.notNull(onError, "onError consumer must not be null"); + Disposable disposable = credentials().subscribe(onNext, onError); + return disposable::dispose; + } + /** * Extension to {@link RedisCredentialsProvider} that resolves credentials immediately without the need to defer the * credential resolution. diff --git a/src/main/java/io/lettuce/core/Subscription.java b/src/main/java/io/lettuce/core/Subscription.java new file mode 100644 index 0000000000..6f84990d93 --- /dev/null +++ b/src/main/java/io/lettuce/core/Subscription.java @@ -0,0 +1,28 @@ +/* + * Copyright (c) 2026-Present, Redis Ltd. All rights reserved. + * SPDX-License-Identifier: MIT + */ +package io.lettuce.core; + +import java.io.Closeable; + +/** + * Handle to a callback subscription on a Lettuce streaming SPI, such as + * {@link io.lettuce.core.RedisCredentialsProvider#subscribeToCredentials(java.util.function.Consumer, java.util.function.Consumer)} + * or {@link io.lettuce.core.event.EventBus}. Closing the subscription stops delivery of further values to the registered + * callback. + *

+ * This is not related to {@code org.reactivestreams.Subscription} or to Redis Pub/Sub channel subscriptions. + * + * @author Aleksandar Todorov + * @since 7.7 + */ +public interface Subscription extends Closeable { + + /** + * Stop delivering to the registered callback. Idempotent; calling it more than once has no further effect and never throws. + */ + @Override + void close(); + +} diff --git a/src/test/java/io/lettuce/core/RedisCredentialsProviderUnitTests.java b/src/test/java/io/lettuce/core/RedisCredentialsProviderUnitTests.java new file mode 100644 index 0000000000..18295b88e3 --- /dev/null +++ b/src/test/java/io/lettuce/core/RedisCredentialsProviderUnitTests.java @@ -0,0 +1,75 @@ +/* + * Copyright (c) 2026-Present, Redis Ltd. All rights reserved. + * SPDX-License-Identifier: MIT + */ +package io.lettuce.core; + +import static io.lettuce.TestTags.UNIT_TEST; +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; + +/** + * Unit tests for the 7.x forward-compatible additions to {@link RedisCredentialsProvider}: {@code resolveCredentialsAsync()} + * and {@code subscribeToCredentials(...)}. + * + * @author Aleksandar Todorov + */ +@SuppressWarnings("deprecation") +@Tag(UNIT_TEST) +class RedisCredentialsProviderUnitTests { + + private final RedisCredentials creds = RedisCredentials.just("user", "pass".toCharArray()); + + @Test + void resolveCredentialsAsyncBridgesResolveCredentials() throws Exception { + + RedisCredentialsProvider provider = () -> Mono.just(creds); + + assertThat(provider.resolveCredentialsAsync().toCompletableFuture().get()).isSameAs(creds); + } + + @Test + void subscribeToCredentialsDeliversAndCloseStops() { + + Sinks.Many sink = Sinks.many().multicast().directBestEffort(); + RedisCredentialsProvider provider = new RedisCredentialsProvider() { + + @Override + public Mono resolveCredentials() { + return sink.asFlux().next(); + } + + @Override + public boolean supportsStreaming() { + return true; + } + + @Override + public Flux credentials() { + return sink.asFlux(); + } + + }; + + List received = new ArrayList<>(); + Subscription subscription = provider.subscribeToCredentials(received::add, t -> { + }); + + sink.tryEmitNext(creds); + assertThat(received).containsExactly(creds); + + subscription.close(); + sink.tryEmitNext(RedisCredentials.just("user2", "pass2".toCharArray())); + assertThat(received).containsExactly(creds); // nothing delivered after close + } + +}