From f558dbb1f28d4cf972af4be3de92ec2cd1fd9cc0 Mon Sep 17 00:00:00 2001 From: "aleksandar.todorov" Date: Thu, 9 Jul 2026 11:57:13 +0300 Subject: [PATCH] Make clean transition from reactive to async credentials provider --- .../TokenBasedRedisCredentialsProvider.java | 11 ++--- .../core/RedisAuthenticationHandler.java | 10 ++--- .../core/RedisCredentialsProvider.java | 26 ++++-------- .../java/io/lettuce/core/RedisHandshake.java | 6 ++- src/main/java/io/lettuce/core/RedisURI.java | 2 +- .../core/StaticCredentialsProvider.java | 2 +- .../java/io/lettuce/core/Subscription.java | 28 +++++++++++++ ...faultAzureCredentialsIntegrationTests.java | 3 +- .../authx/EntraIdIntegrationTests.java | 5 +-- ...okenBasedRedisCredentialsProviderTest.java | 22 +++++----- .../ConnectionCommandIntegrationTests.java | 4 +- .../MyStreamingRedisCredentialsProvider.java | 4 +- .../lettuce/core/RedisHandshakeUnitTests.java | 2 +- .../core/RedisURIBuilderUnitTests.java | 42 +++++++++++-------- .../io/lettuce/core/RedisURIUnitTests.java | 10 ++--- .../cluster/RedisClusterURIUtilUnitTests.java | 6 +-- 16 files changed, 104 insertions(+), 79 deletions(-) create mode 100644 src/main/java/io/lettuce/core/Subscription.java diff --git a/src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java b/src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java index 014dd85d43..5c2043dab4 100644 --- a/src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java +++ b/src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java @@ -8,6 +8,7 @@ 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; @@ -52,7 +53,7 @@ public class TokenBasedRedisCredentialsProvider implements RedisCredentialsProvi 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 resolveCredentials() { + public CompletionStage resolveCredentialsAsync() { CompletableFuture result = new CompletableFuture<>(); credentialsFutureRef.get().whenComplete((creds, throwable) -> { if (throwable != null) { @@ -218,7 +219,7 @@ public CompletionStage resolveCredentials() { } @Override - public CredentialsSubscription subscribeToCredentials(Consumer onNext, Consumer onError) { + public Subscription subscribeToCredentials(Consumer onNext, Consumer onError) { if (isClosed) { throw new IllegalStateException("Credentials provider closed"); } @@ -280,7 +281,7 @@ public void close() { *
    *
  • 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.
    • - *
  • Continuations chained off {@link #resolveCredentials()} run on the renewal thread when the initial future + *
  • Continuations chained off {@link #resolveCredentialsAsync()} run on the renewal thread when the initial future * completes.
    • *
  • Replay deliveries to a newly subscribing consumer run on the subscribing thread.
    • *
@@ -323,7 +324,7 @@ public static TokenBasedRedisCredentialsProvider create(TokenAuthConfig tokenAut *
    *
  • 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.
    • - *
  • Continuations chained off {@link #resolveCredentials()} run on the renewal thread when the initial future + *
  • Continuations chained off {@link #resolveCredentialsAsync()} run on the renewal thread when the initial future * completes.
    • *
  • Replay deliveries to a newly subscribing consumer run on the subscribing thread.
    • *
diff --git a/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java b/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java index d0e3c47da6..5fe9dac919 100644 --- a/src/main/java/io/lettuce/core/RedisAuthenticationHandler.java +++ b/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; @@ -48,7 +47,7 @@ public class RedisAuthenticationHandler { private final RedisCredentialsProvider credentialsProvider; - private final AtomicReference credentialsSubscription = new AtomicReference<>(); + private final AtomicReference credentialsSubscription = new AtomicReference<>(); private final Boolean isPubSubConnection; @@ -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(); diff --git a/src/main/java/io/lettuce/core/RedisCredentialsProvider.java b/src/main/java/io/lettuce/core/RedisCredentialsProvider.java index 86dc509b96..b75314f3d9 100644 --- a/src/main/java/io/lettuce/core/RedisCredentialsProvider.java +++ b/src/main/java/io/lettuce/core/RedisCredentialsProvider.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; @@ -22,26 +21,17 @@ @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(); - - } - /** * 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. + * 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 resolveCredentials(); + CompletionStage resolveCredentialsAsync(); /** * Creates a new {@link RedisCredentialsProvider} from a given {@link Supplier}. @@ -101,10 +91,10 @@ 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 onNext, Consumer onError) { + default Subscription subscribeToCredentials(Consumer onNext, Consumer onError) { throw new UnsupportedOperationException("Streaming credentials are not supported by this provider."); } @@ -116,7 +106,7 @@ default CredentialsSubscription subscribeToCredentials(Consumer resolveCredentials() { + default CompletionStage resolveCredentialsAsync() { try { RedisCredentials credentials = resolveCredentialsNow(); if (credentials == null) { diff --git a/src/main/java/io/lettuce/core/RedisHandshake.java b/src/main/java/io/lettuce/core/RedisHandshake.java index bd221e8542..89e6cd4dc7 100644 --- a/src/main/java/io/lettuce/core/RedisHandshake.java +++ b/src/main/java/io/lettuce/core/RedisHandshake.java @@ -210,7 +210,8 @@ private CompletableFuture initiateHandshakeResp2(Channel channel, RedisCreden ((RedisCredentialsProvider.ImmediateRedisCredentialsProvider) credentialsProvider).resolveCredentialsNow()); } - CompletableFuture credentialsFuture = credentialsProvider.resolveCredentials().toCompletableFuture(); + CompletableFuture credentialsFuture = credentialsProvider.resolveCredentialsAsync() + .toCompletableFuture(); return credentialsFuture.thenComposeAsync(credentials -> dispatchAuthOrPing(channel, credentials)); } @@ -243,7 +244,8 @@ private CompletionStage> initiateHandshakeResp3(Channel chan ((RedisCredentialsProvider.ImmediateRedisCredentialsProvider) credentialsProvider).resolveCredentialsNow()); } - CompletableFuture credentialsFuture = credentialsProvider.resolveCredentials().toCompletableFuture(); + CompletableFuture credentialsFuture = credentialsProvider.resolveCredentialsAsync() + .toCompletableFuture(); return credentialsFuture.thenComposeAsync(credentials -> dispatchHello(channel, credentials)); } diff --git a/src/main/java/io/lettuce/core/RedisURI.java b/src/main/java/io/lettuce/core/RedisURI.java index 0d52688aaf..3e5e412d0f 100644 --- a/src/main/java/io/lettuce/core/RedisURI.java +++ b/src/main/java/io/lettuce/core/RedisURI.java @@ -976,7 +976,7 @@ private String getAuthority(String scheme, boolean maskCredentials) { // would get asterix for each character of the password. RedisCredentials creds; try { - creds = credentialsProvider.resolveCredentials().toCompletableFuture().join(); + creds = credentialsProvider.resolveCredentialsAsync().toCompletableFuture().join(); } catch (Exception e) { throw Exceptions.bubble(e); } diff --git a/src/main/java/io/lettuce/core/StaticCredentialsProvider.java b/src/main/java/io/lettuce/core/StaticCredentialsProvider.java index 507e1e0d4a..931eab2f36 100644 --- a/src/main/java/io/lettuce/core/StaticCredentialsProvider.java +++ b/src/main/java/io/lettuce/core/StaticCredentialsProvider.java @@ -44,7 +44,7 @@ public StaticCredentialsProvider(RedisCredentials credentials) { } @Override - public CompletionStage resolveCredentials() { + public CompletionStage resolveCredentialsAsync() { return future; } 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/authx/DefaultAzureCredentialsIntegrationTests.java b/src/test/java/io/lettuce/authx/DefaultAzureCredentialsIntegrationTests.java index 2f0624ede5..c8083d4316 100644 --- a/src/test/java/io/lettuce/authx/DefaultAzureCredentialsIntegrationTests.java +++ b/src/test/java/io/lettuce/authx/DefaultAzureCredentialsIntegrationTests.java @@ -79,7 +79,8 @@ public void cleanUp() { @Test public void azureTokenAuthWithDefaultAzureCredentials() throws ExecutionException, InterruptedException, TimeoutException { - RedisCredentials credentials = credentialsProvider.resolveCredentials().toCompletableFuture().get(5, TimeUnit.SECONDS); + RedisCredentials credentials = credentialsProvider.resolveCredentialsAsync().toCompletableFuture().get(5, + TimeUnit.SECONDS); assertThat(credentials).isNotNull(); String key = UUID.randomUUID().toString(); diff --git a/src/test/java/io/lettuce/authx/EntraIdIntegrationTests.java b/src/test/java/io/lettuce/authx/EntraIdIntegrationTests.java index 5064990cdb..3e552a379b 100644 --- a/src/test/java/io/lettuce/authx/EntraIdIntegrationTests.java +++ b/src/test/java/io/lettuce/authx/EntraIdIntegrationTests.java @@ -1,7 +1,6 @@ package io.lettuce.authx; import io.lettuce.core.*; -import io.lettuce.core.RedisCredentialsProvider.CredentialsSubscription; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.async.RedisAsyncCommands; import io.lettuce.core.api.reactive.RedisReactiveCommands; @@ -116,7 +115,7 @@ public void renewalDuringOperationsTest() throws InterruptedException { commandThread.start(); CountDownLatch latch = new CountDownLatch(10); // Wait for at least 10 token renewalss - CredentialsSubscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> { + Subscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> { }); try { assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); // Wait to reach 10 renewals @@ -150,7 +149,7 @@ public void renewalDuringPubSubOperationsTest() throws InterruptedException { pubsubThread.start(); CountDownLatch latch = new CountDownLatch(10); - CredentialsSubscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> { + Subscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> { }); try { assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); // Wait for at least 10 token renewals diff --git a/src/test/java/io/lettuce/authx/TokenBasedRedisCredentialsProviderTest.java b/src/test/java/io/lettuce/authx/TokenBasedRedisCredentialsProviderTest.java index a71e1d3e7d..ac5eb66161 100644 --- a/src/test/java/io/lettuce/authx/TokenBasedRedisCredentialsProviderTest.java +++ b/src/test/java/io/lettuce/authx/TokenBasedRedisCredentialsProviderTest.java @@ -2,7 +2,7 @@ import io.lettuce.TestTags; import io.lettuce.core.RedisCredentials; -import io.lettuce.core.RedisCredentialsProvider.CredentialsSubscription; +import io.lettuce.core.Subscription; import io.lettuce.core.TestTokenManager; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Tag; @@ -44,7 +44,7 @@ public void setUp() { public void shouldReturnPreviouslyEmittedTokenWhenResolved() { tokenManager.emitToken(testToken("test-user", "token-1")); - Mono credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentials()); + Mono credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync()); StepVerifier.create(credentials).assertNext(actual -> { assertThat(actual.getUsername()).isEqualTo("test-user"); @@ -57,7 +57,7 @@ public void shouldReturnLatestEmittedTokenWhenResolved() { tokenManager.emitToken(testToken("test-user", "token-2")); tokenManager.emitToken(testToken("test-user", "token-3")); // Latest token - Mono credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentials()); + Mono credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync()); StepVerifier.create(credentials).assertNext(actual -> { assertThat(actual.getUsername()).isEqualTo("test-user"); @@ -71,7 +71,7 @@ public void shouldReturnTokenEmittedBeforeSubscription() { tokenManager.emitToken(testToken("test-user", "token-1")); // Test resolveCredentials - Mono credentials1 = Mono.fromCompletionStage(credentialsProvider.resolveCredentials()); + Mono credentials1 = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync()); StepVerifier.create(credentials1).assertNext(actual -> { assertThat(actual.getUsername()).isEqualTo("test-user"); @@ -81,7 +81,7 @@ public void shouldReturnTokenEmittedBeforeSubscription() { // Emit second token and subscribe another tokenManager.emitToken(testToken("test-user", "token-2")); tokenManager.emitToken(testToken("test-user", "token-3")); - Mono credentials2 = Mono.fromCompletionStage(credentialsProvider.resolveCredentials()); + Mono credentials2 = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync()); StepVerifier.create(credentials2).assertNext(actual -> { assertThat(actual.getUsername()).isEqualTo("test-user"); assertThat(new String(actual.getPassword())).isEqualTo("token-3"); @@ -90,7 +90,7 @@ public void shouldReturnTokenEmittedBeforeSubscription() { @Test public void shouldWaitForAndReturnTokenWhenEmittedLater() { - Mono result = Mono.fromCompletionStage(credentialsProvider.resolveCredentials()); + Mono result = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync()); tokenManager.emitTokenWithDelay(testToken("test-user", "delayed-token"), 100); // Emit token after 100ms StepVerifier.create(result) @@ -104,12 +104,12 @@ public void shouldStopDeliveringToSubscribersOnClose() throws InterruptedExcepti List received2 = new CopyOnWriteArrayList<>(); CountDownLatch firstTokenLatch = new CountDownLatch(2); - CredentialsSubscription sub1 = credentialsProvider.subscribeToCredentials(c -> { + Subscription sub1 = credentialsProvider.subscribeToCredentials(c -> { received1.add(c); firstTokenLatch.countDown(); }, t -> { }); - CredentialsSubscription sub2 = credentialsProvider.subscribeToCredentials(c -> { + Subscription sub2 = credentialsProvider.subscribeToCredentials(c -> { received2.add(c); firstTokenLatch.countDown(); }, t -> { @@ -140,7 +140,7 @@ public void shouldPropagateMultipleTokensOnStream() throws InterruptedException List received = new CopyOnWriteArrayList<>(); CountDownLatch latch = new CountDownLatch(2); - CredentialsSubscription sub = credentialsProvider.subscribeToCredentials(c -> { + Subscription sub = credentialsProvider.subscribeToCredentials(c -> { received.add(c); latch.countDown(); }, t -> { @@ -166,7 +166,7 @@ public void shouldReplayLatestTokenToNewSubscriber() throws InterruptedException AtomicReference received = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); - CredentialsSubscription sub = credentialsProvider.subscribeToCredentials(c -> { + Subscription sub = credentialsProvider.subscribeToCredentials(c -> { received.set(c); latch.countDown(); }, t -> { @@ -187,7 +187,7 @@ public void shouldPropagateTokenRequestErrorsToSubscribers() throws InterruptedE CountDownLatch tokensLatch = new CountDownLatch(2); CountDownLatch errorLatch = new CountDownLatch(1); - CredentialsSubscription sub = credentialsProvider.subscribeToCredentials(c -> { + Subscription sub = credentialsProvider.subscribeToCredentials(c -> { received.add(c); tokensLatch.countDown(); }, t -> { diff --git a/src/test/java/io/lettuce/core/ConnectionCommandIntegrationTests.java b/src/test/java/io/lettuce/core/ConnectionCommandIntegrationTests.java index c1c34cfe58..1146bd7678 100644 --- a/src/test/java/io/lettuce/core/ConnectionCommandIntegrationTests.java +++ b/src/test/java/io/lettuce/core/ConnectionCommandIntegrationTests.java @@ -285,8 +285,8 @@ void authInvalidPassword() { } catch (RedisException e) { assertThat(e.getMessage()).startsWith("ERR").contains("AUTH"); StatefulRedisConnectionImpl connectionImpl = (StatefulRedisConnectionImpl) connection; - assertThat(connectionImpl.getConnectionState().getCredentialsProvider().resolveCredentials().toCompletableFuture() - .join().getPassword()).isNull(); + assertThat(connectionImpl.getConnectionState().getCredentialsProvider().resolveCredentialsAsync() + .toCompletableFuture().join().getPassword()).isNull(); } finally { connection.close(); } diff --git a/src/test/java/io/lettuce/core/MyStreamingRedisCredentialsProvider.java b/src/test/java/io/lettuce/core/MyStreamingRedisCredentialsProvider.java index 38dd1cc915..48988e00a8 100644 --- a/src/test/java/io/lettuce/core/MyStreamingRedisCredentialsProvider.java +++ b/src/test/java/io/lettuce/core/MyStreamingRedisCredentialsProvider.java @@ -30,12 +30,12 @@ public boolean supportsStreaming() { } @Override - public CompletionStage resolveCredentials() { + public CompletionStage resolveCredentialsAsync() { return credentialsFutureRef.get(); } @Override - public CredentialsSubscription subscribeToCredentials(Consumer onNext, Consumer onError) { + public Subscription subscribeToCredentials(Consumer onNext, Consumer onError) { LettuceAssert.notNull(onNext, "onNext consumer must not be null"); LettuceAssert.notNull(onError, "onError consumer must not be null"); Listener listener = new Listener(onNext, onError); diff --git a/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java b/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java index 926ffb1b4e..dbd128a852 100644 --- a/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java +++ b/src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java @@ -359,7 +359,7 @@ static class DelayedRedisCredentialsProvider implements RedisCredentialsProvider private final Sinks.One credentialsSink = Sinks.one(); @Override - public CompletionStage resolveCredentials() { + public CompletionStage resolveCredentialsAsync() { return credentialsSink.asMono().toFuture(); } diff --git a/src/test/java/io/lettuce/core/RedisURIBuilderUnitTests.java b/src/test/java/io/lettuce/core/RedisURIBuilderUnitTests.java index 1c2ee31b65..34aabc07cb 100644 --- a/src/test/java/io/lettuce/core/RedisURIBuilderUnitTests.java +++ b/src/test/java/io/lettuce/core/RedisURIBuilderUnitTests.java @@ -171,7 +171,7 @@ void redisFromUrl() { assertThat(result.getSentinels()).isEmpty(); assertThat(result.getHost()).isEqualTo("localhost"); assertThat(result.getPort()).isEqualTo(RedisURI.DEFAULT_REDIS_PORT); - StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -183,14 +183,14 @@ void redisFromUrl() { @Test void redisFromUrlNoPassword() { RedisURI redisURI = RedisURI.create("redis://localhost:1234/5"); - StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isNull(); }).verifyComplete(); redisURI = RedisURI.create("redis://h:@localhost.com:14589"); - StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isNull(); @@ -201,7 +201,7 @@ void redisFromUrlNoPassword() { void redisFromUrlPassword() { RedisURI redisURI = RedisURI.create("redis://h:password@localhost.com:14589"); - StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isEqualTo("h"); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -236,7 +236,7 @@ void redisSslFromUrl() { assertThat(result.getSentinels()).isEmpty(); assertThat(result.getHost()).isEqualTo("localhost"); assertThat(result.getPort()).isEqualTo(RedisURI.DEFAULT_REDIS_PORT); - StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -254,7 +254,7 @@ void redisSentinelFromUrl() { assertThat(result.getPort()).isEqualTo(RedisURI.DEFAULT_REDIS_PORT); assertThat(result.getSentinelMasterId()).isEqualTo("master"); assertThat(result.toString()).contains("master"); - StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -266,7 +266,7 @@ void redisSentinelFromUrl() { assertThat(result.getHost()).isNull(); assertThat(result.getPort()).isEqualTo(RedisURI.DEFAULT_REDIS_PORT); assertThat(result.getSentinelMasterId()).isEqualTo("master"); - StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -291,7 +291,7 @@ void withAuthenticatedSentinel() { RedisURI result = RedisURI.Builder.sentinel("host", 1234, "master", "foo").build(); RedisURI sentinel = result.getSentinels().get(0); - StepVerifier.create(Mono.fromCompletionStage(sentinel.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(sentinel.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("foo".toCharArray()); @@ -309,7 +309,7 @@ void withTlsSentinel() { assertThat(sentinel.isStartTls()).isTrue(); assertThat(sentinel.isVerifyPeer()).isFalse(); - StepVerifier.create(Mono.fromCompletionStage(sentinel.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(sentinel.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("foo".toCharArray()); @@ -324,14 +324,16 @@ void withAuthenticatedSentinelUri() { RedisURI result = RedisURI.Builder.sentinel("host", 1234, "master").withSentinel(sentinel).build(); StepVerifier - .create(Mono.fromCompletionStage(result.getSentinels().get(0).getCredentialsProvider().resolveCredentials())) + .create(Mono + .fromCompletionStage(result.getSentinels().get(0).getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isNull(); }).verifyComplete(); StepVerifier - .create(Mono.fromCompletionStage(result.getSentinels().get(1).getCredentialsProvider().resolveCredentials())) + .create(Mono + .fromCompletionStage(result.getSentinels().get(1).getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("bar".toCharArray()); @@ -344,14 +346,16 @@ void withAuthenticatedSentinelWithSentinel() { RedisURI result = RedisURI.Builder.sentinel("host", 1234, "master", "foo").withSentinel("bar").build(); StepVerifier - .create(Mono.fromCompletionStage(result.getSentinels().get(0).getCredentialsProvider().resolveCredentials())) + .create(Mono + .fromCompletionStage(result.getSentinels().get(0).getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("foo".toCharArray()); }).verifyComplete(); StepVerifier - .create(Mono.fromCompletionStage(result.getSentinels().get(1).getCredentialsProvider().resolveCredentials())) + .create(Mono + .fromCompletionStage(result.getSentinels().get(1).getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isNull(); @@ -361,14 +365,16 @@ void withAuthenticatedSentinelWithSentinel() { .withSentinel("bar", 1234, "baz").build(); StepVerifier - .create(Mono.fromCompletionStage(result.getSentinels().get(0).getCredentialsProvider().resolveCredentials())) + .create(Mono + .fromCompletionStage(result.getSentinels().get(0).getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isNull(); }).verifyComplete(); StepVerifier - .create(Mono.fromCompletionStage(result.getSentinels().get(1).getCredentialsProvider().resolveCredentials())) + .create(Mono + .fromCompletionStage(result.getSentinels().get(1).getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("baz".toCharArray()); @@ -425,7 +431,7 @@ void redisSocket() throws IOException { assertThat(result.getPort()).isEqualTo(RedisURI.DEFAULT_REDIS_PORT); assertThat(result.isSsl()).isFalse(); - StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isNull(); @@ -444,7 +450,7 @@ void redisSocketWithPassword() throws IOException { assertThat(result.getPort()).isEqualTo(RedisURI.DEFAULT_REDIS_PORT); assertThat(result.isSsl()).isFalse(); - StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(result.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -497,7 +503,7 @@ void shouldInitializeBuilder() { assertThat(target.isSsl()).isEqualTo(source.isSsl()); assertThat(target.isVerifyPeer()).isEqualTo(source.isVerifyPeer()); - StepVerifier.create(Mono.fromCompletionStage(target.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(target.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("baz".toCharArray()); diff --git a/src/test/java/io/lettuce/core/RedisURIUnitTests.java b/src/test/java/io/lettuce/core/RedisURIUnitTests.java index 189ccdfad6..8c5aa2b3f5 100644 --- a/src/test/java/io/lettuce/core/RedisURIUnitTests.java +++ b/src/test/java/io/lettuce/core/RedisURIUnitTests.java @@ -284,7 +284,7 @@ void escapeCharacterParsingTest() throws UnsupportedEncodingException { String uri = "redis-sentinel://" + translatedPassword + "@h1:1234,h2:1234,h3:1234/0?sentinelMasterId=masterId"; RedisURI redisURI = RedisURI.create(uri); assertThat(redisURI.getSentinels().get(0).getHost()).isEqualTo("h1"); - StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo(password.toCharArray()); @@ -294,7 +294,7 @@ void escapeCharacterParsingTest() throws UnsupportedEncodingException { uri = "redis://" + translatedPassword + "@h1:1234/0"; redisURI = RedisURI.create(uri); assertThat(redisURI.getHost()).isEqualTo("h1"); - StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(redisURI.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo(password.toCharArray()); @@ -427,7 +427,7 @@ void shouldApplyAuthentication() { RedisURI target = new RedisURI(); target.applyAuthentication(source); - StepVerifier.create(Mono.fromCompletionStage(target.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(target.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isEqualTo("foo"); assertThat(credentials.getPassword()).isEqualTo("bar".toCharArray()); @@ -436,7 +436,7 @@ void shouldApplyAuthentication() { source.setCredentialsProvider(new StaticCredentialsProvider(null, "bar".toCharArray())); target.applyAuthentication(source); - StepVerifier.create(Mono.fromCompletionStage(target.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(target.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("bar".toCharArray()); @@ -451,7 +451,7 @@ void shouldApplyAuthentication() { RedisURI targetCp = new RedisURI(); targetCp.applyAuthentication(sourceCp); - StepVerifier.create(Mono.fromCompletionStage(targetCp.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(targetCp.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isEqualTo("suppliedUsername"); assertThat(credentials.getPassword()).isEqualTo("suppliedPassword".toCharArray()); diff --git a/src/test/java/io/lettuce/core/cluster/RedisClusterURIUtilUnitTests.java b/src/test/java/io/lettuce/core/cluster/RedisClusterURIUtilUnitTests.java index 58bef69fcf..6a12545698 100644 --- a/src/test/java/io/lettuce/core/cluster/RedisClusterURIUtilUnitTests.java +++ b/src/test/java/io/lettuce/core/cluster/RedisClusterURIUtilUnitTests.java @@ -75,7 +75,7 @@ void testSslWithPasswordSingleHost() { assertThat(host1.isStartTls()).isTrue(); assertThat(host1.getHost()).isEqualTo("host1"); assertThat(host1.getPort()).isEqualTo(6379); - StepVerifier.create(Mono.fromCompletionStage(host1.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(host1.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -94,7 +94,7 @@ void testSslWithPasswordMultipleHosts() { assertThat(host1.isStartTls()).isTrue(); assertThat(host1.getHost()).isEqualTo("host1"); assertThat(host1.getPort()).isEqualTo(6379); - StepVerifier.create(Mono.fromCompletionStage(host1.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(host1.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray()); @@ -105,7 +105,7 @@ void testSslWithPasswordMultipleHosts() { assertThat(host2.isStartTls()).isTrue(); assertThat(host2.getHost()).isEqualTo("host2"); assertThat(host2.getPort()).isEqualTo(6380); - StepVerifier.create(Mono.fromCompletionStage(host2.getCredentialsProvider().resolveCredentials())) + StepVerifier.create(Mono.fromCompletionStage(host2.getCredentialsProvider().resolveCredentialsAsync())) .assertNext(credentials -> { assertThat(credentials.getUsername()).isNull(); assertThat(credentials.getPassword()).isEqualTo("password".toCharArray());