Rebase on PR #3068

# Conflicts:
#	src/test/java/io/lettuce/core/AuthenticationIntegrationTests.java

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

@@ -35,10 +35,14 @@ private void initializeTokenManager() {
 
             @Override
             public void onTokenRenewed(Token token) {
-                String username = token.tryGet("oid");
-                char[] pass = token.getValue().toCharArray();
-                RedisCredentials credentials = RedisCredentials.just(username, pass);
-                credentialsSink.tryEmitNext(credentials);
+                try {
+                    String username = token.tryGet("oid");
+                    char[] pass = token.getValue().toCharArray();
+                    RedisCredentials credentials = RedisCredentials.just(username, pass);
+                    credentialsSink.tryEmitNext(credentials);
+                } catch (Exception e) {
+                    credentialsSink.emitError(e, Sinks.EmitFailureHandler.FAIL_FAST);
+                }
             }
 
             @Override

src/test/java/io/lettuce/core/AuthenticationIntegrationTests.java

@@ -5,13 +5,10 @@
 
 import javax.inject.Inject;
 
-import io.lettuce.authx.TokenBasedRedisCredentialsProvider;
 import io.lettuce.authx.TokenBasedRedisCredentialsProvider;
 import io.lettuce.core.event.command.CommandListener;
 import io.lettuce.core.event.command.CommandSucceededEvent;
 import io.lettuce.core.protocol.RedisCommand;
-import io.lettuce.test.Delay;
-import io.lettuce.test.Delay;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
@@ -27,11 +24,13 @@
 import io.lettuce.test.WithPassword;
 import io.lettuce.test.condition.EnabledOnCommand;
 import io.lettuce.test.settings.TestSettings;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import redis.clients.authentication.core.SimpleToken;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 /**
@@ -126,6 +125,43 @@ void streamingCredentialProvider(RedisClient client) {
                 client.getOptions().mutate().reauthenticateBehavior(ClientOptions.ReauthenticateBehavior.DEFAULT).build());
     }
 
+    @Test
+    @Inject
+    void tokenBasedCredentialProvider(RedisClient client) {
+
+        TestCommandListener listener = new TestCommandListener();
+        client.addListener(listener);
+        client.setOptions(client.getOptions().mutate()
+                .reauthenticateBehavior(ClientOptions.ReauthenticateBehavior.ON_NEW_CREDENTIALS).build());
+
+        TestTokenManager tokenManager = new TestTokenManager(null, null);
+        TokenBasedRedisCredentialsProvider credentialsProvider = new TokenBasedRedisCredentialsProvider(tokenManager);
+
+        // Build RedisURI with streaming credentials provider
+        RedisURI uri = RedisURI.builder().withHost(TestSettings.host()).withPort(TestSettings.port())
+                .withClientName("streaming_cred_test").withAuthentication(credentialsProvider)
+                .withTimeout(Duration.ofSeconds(5)).build();
+        tokenManager.emitToken(testToken(TestSettings.username(), TestSettings.password().toString().toCharArray()));
+
+        StatefulRedisConnection<String, String> connection = client.connect(StringCodec.UTF8, uri);
+        assertThat(connection.sync().aclWhoami()).isEqualTo(TestSettings.username());
+
+        // rotate the credentials
+        tokenManager.emitToken(testToken("steave", "foobared".toCharArray()));
+
+        Awaitility.await().atMost(Duration.ofSeconds(1)).until(() -> listener.succeeded.stream()
+                .anyMatch(command -> isAuthCommandWithCredentials(command, "steave", "foobared".toCharArray())));
+
+        // verify that the connection is re-authenticated with the new user credentials
+        assertThat(connection.sync().aclWhoami()).isEqualTo("steave");
+
+        credentialsProvider.shutdown();
+        connection.close();
+        client.removeListener(listener);
+        client.setOptions(
+                client.getOptions().mutate().reauthenticateBehavior(ClientOptions.ReauthenticateBehavior.DEFAULT).build());
+    }
+
     static class TestCommandListener implements CommandListener {
 
         final List<RedisCommand<?, ?, ?>> succeeded = new ArrayList<>();
@@ -147,52 +183,9 @@ private boolean isAuthCommandWithCredentials(RedisCommand<?, ?, ?> command, Stri
         return false;
     }
 
-}
-
-@Test
-@Inject
-void tokenBasedCredentialProvider(RedisClient client) {
-
-    ClientOptions clientOptions = ClientOptions.builder()
-            .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS).build();
-    client.setOptions(clientOptions);
-    // Connection used to simulate test user credential rotation
-    StatefulRedisConnection<String, String> defaultConnection = client.connect();
-
-    String testUser = "streaming_cred_test_user";
-    char[] testPassword1 = "token_1".toCharArray();
-    char[] testPassword2 = "token_2".toCharArray();
-
-    TestTokenManager tokenManager = new TestTokenManager(null, null);
-
-    // streaming credentials provider that emits redis credentials which will trigger connection re-authentication
-    // token manager is used to emit updated credentials
-    TokenBasedRedisCredentialsProvider credentialsProvider = new TokenBasedRedisCredentialsProvider(tokenManager);
-
-    RedisURI uri = RedisURI.builder().withTimeout(Duration.ofSeconds(1)).withClientName("streaming_cred_test")
-            .withHost(TestSettings.host()).withPort(TestSettings.port()).withAuthentication(credentialsProvider).build();
-
-    // create test user with initial credentials set to 'testPassword1'
-    createTestUser(defaultConnection, testUser, testPassword1);
-    tokenManager.emitToken(testToken(testUser, testPassword1));
-
-    StatefulRedisConnection<String, String> connection = client.connect(StringCodec.UTF8, uri);
-    assertThat(connection.sync().aclWhoami()).isEqualTo(testUser);
-
-    // update test user credentials in Redis server (password changed to testPassword2)
-    // then emit updated credentials trough streaming credentials provider
-    // and trigger re-connect to force re-authentication
-    // updated credentials should be used for re-authentication
-    updateTestUser(defaultConnection, testUser, testPassword2);
-    tokenManager.emitToken(testToken(testUser, testPassword2));
-    connection.sync().quit();
-
-    Delay.delay(Duration.ofMillis(100));
-    assertThat(connection.sync().ping()).isEqualTo("PONG");
-
-    String res = connection.sync().aclWhoami();
-    assertThat(res).isEqualTo(testUser);
+    private SimpleToken testToken(String username, char[] password) {
+        return new SimpleToken(String.valueOf(password), Instant.now().plusMillis(500).toEpochMilli(),
+                Instant.now().toEpochMilli(), Collections.singletonMap("oid", username));
+    }
 
-    defaultConnection.close();
-    connection.close();
 }

src/test/java/io/lettuce/core/cluster/RedisClusterStreamingCredentialsProviderlIntegrationTests.java

@@ -140,7 +140,6 @@ void nodeSelectionApiShouldWork() {
     @Test
     void shouldPerformNodeConnectionReauth() {
         ClusterClientOptions origClientOptions = redisClient.getClusterClientOptions();
-        origClientOptions.mutate().reauthenticateBehavior(ClientOptions.ReauthenticateBehavior.ON_NEW_CREDENTIALS).build();
         redisClient.setOptions(origClientOptions.mutate()
                 .reauthenticateBehavior(ClientOptions.ReauthenticateBehavior.ON_NEW_CREDENTIALS).build());
 

src/test/java/io/lettuce/examples/TokenBasedAuthExample.java

@@ -1,20 +1,29 @@
 package io.lettuce.examples;
 
+import com.microsoft.aad.msal4j.ClientCredentialFactory;
+import com.microsoft.aad.msal4j.ClientCredentialParameters;
+import com.microsoft.aad.msal4j.ConfidentialClientApplication;
+import com.microsoft.aad.msal4j.IAuthenticationResult;
+import com.microsoft.aad.msal4j.IClientSecret;
 import io.lettuce.authx.TokenBasedRedisCredentialsProvider;
 import io.lettuce.core.ClientOptions;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.SocketOptions;
-import io.lettuce.core.TimeoutOptions;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.codec.StringCodec;
 import redis.clients.authentication.core.IdentityProviderConfig;
 import redis.clients.authentication.core.TokenAuthConfig;
 import redis.clients.authentication.entraid.EntraIDTokenAuthConfigBuilder;
 
+import java.net.MalformedURLException;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import static org.assertj.core.api.Assertions.assertThat;
 
 public class TokenBasedAuthExample {
 
@@ -24,11 +33,34 @@ public static void main(String[] args) {
         Set<String> scopes = Collections.singleton("https://redis.azure.com/.default");
 
         String User1_clientId = System.getenv("USER1_CLIENT_ID");
+        String User1_objectid = System.getenv("USER1_OBJECT_ID");
         String User1_secret = System.getenv("USER1_SECRET");
 
         String User2_clientId = System.getenv("USER2_CLIENT_ID");
         String User2_secret = System.getenv("USER2_SECRET");
 
+        try {
+            IClientSecret cred = ClientCredentialFactory.createFromSecret(User1_secret);
+            ConfidentialClientApplication app = ConfidentialClientApplication.builder(User1_clientId, cred).authority(authority)
+                    .build();
+            ClientCredentialParameters params = ClientCredentialParameters.builder(scopes).skipCache(true).build();
+            Future<IAuthenticationResult> tokenRequest1 = app.acquireToken(params);
+            IAuthenticationResult t1 = tokenRequest1.get();
+            Future<IAuthenticationResult> tokenRequest2 = app.acquireToken(params);
+            IAuthenticationResult t2 = tokenRequest2.get();
+            System.out.println(t1.accessToken());
+            System.out.println(t2.accessToken());
+            assertThat(t1.accessToken()).isNotEqualTo(t2.accessToken());
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } catch (ExecutionException e) {
+            throw new RuntimeException(e);
+        } catch (MalformedURLException e) {
+            throw new RuntimeException(e);
+        }
+
+        ClientCredentialParameters.builder(scopes).skipCache(true).build();
+
         // User 1
         // from redis-authx-entraind
         IdentityProviderConfig config1 = EntraIDTokenAuthConfigBuilder.builder().authority(authority).clientId(User1_clientId)
@@ -62,7 +94,7 @@ public static void main(String[] args) {
         ClientOptions clientOptions = ClientOptions.builder()
                 .socketOptions(SocketOptions.builder().connectTimeout(Duration.ofSeconds(5)).build())
                 .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
-                .timeoutOptions(TimeoutOptions.enabled(Duration.ofSeconds(1))).build();
+                .reauthenticateBehavior(ClientOptions.ReauthenticateBehavior.ON_NEW_CREDENTIALS).build();
 
         // RedisClient using user1 credentials by default
         RedisClient redisClient = RedisClient.create(redisURI1);