Merge feature/tokenrefreshtoptions to master

eng/jacoco-test-coverage/pom.xml

@@ -134,7 +134,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-identity</artifactId>
-      <version>1.1.0-beta.7</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
+      <version>1.1.0-beta.8</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>

eng/versioning/version_client.txt

@@ -27,7 +27,7 @@ com.azure:azure-data-schemaregistry;1.0.0-beta.2;1.0.0-beta.3
 com.azure:azure-data-schemaregistry-avro;1.0.0-beta.2;1.0.0-beta.3
 com.azure:azure-data-tables;1.0.0-beta.1;1.0.0-beta.1
 com.azure:azure-e2e;1.0.0-beta.1;1.0.0-beta.1
-com.azure:azure-identity;1.0.9;1.1.0-beta.7
+com.azure:azure-identity;1.0.9;1.1.0-beta.8
 com.azure:azure-identity-perf;1.0.0-beta.1;1.0.0-beta.1
 com.azure:azure-messaging-eventhubs;5.1.2;5.2.0-beta.1
 com.azure:azure-messaging-eventhubs-checkpointstore-blob;1.1.2;1.2.0-beta.2
@@ -75,8 +75,9 @@ com.microsoft.azure:spring-data-cosmosdb;3.0.0-beta.1;3.0.0-beta.1
 # Format;
 # unreleased_<groupId>:<artifactId>;dependency-version
 # note: The unreleased dependencies will not be manipulated with the automatic PR creation code.
-
+unreleased_com.azure:azure-core;1.7.0-beta.3
 unreleased_com.azure:azure-core-test;1.4.0-beta.1
+unreleased_com.azure:azure-identity;1.1.0-beta.8
 unreleased_com.azure:azure-messaging-servicebus;7.0.0-beta.4
 unreleased_com.azure:azure-security-keyvault-keys;4.2.0-beta.6
 

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ClaimsBasedSecurityChannelTest.java

@@ -107,8 +107,7 @@ public void teardown() {
     @Test
     public void authorizesSasToken() {
         // Arrange
-        // Subtracting two minutes because the AccessToken does this internally.
-        final Date expectedDate = Date.from(validUntil.minusMinutes(2).toInstant());
+        final Date expectedDate = Date.from(validUntil.toInstant());
         final ClaimsBasedSecurityChannel cbsChannel = new ClaimsBasedSecurityChannel(Mono.just(requestResponseChannel),
             tokenCredential, CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, options);
 

sdk/core/azure-core/src/main/java/com/azure/core/credential/AccessToken.java

@@ -19,7 +19,7 @@ public class AccessToken {
      */
     public AccessToken(String token, OffsetDateTime expiresAt) {
         this.token = token;
-        this.expiresAt = expiresAt.minusMinutes(2); // 2 minutes before token expires
+        this.expiresAt = expiresAt;
     }
 
     /**

sdk/core/azure-core/src/main/java/com/azure/core/credential/SimpleTokenCache.java

@@ -3,53 +3,139 @@
 
 package com.azure.core.credential;
 
-import reactor.core.publisher.FluxSink;
-import reactor.core.publisher.FluxSink.OverflowStrategy;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.ReplayProcessor;
+import reactor.core.publisher.MonoProcessor;
 
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 /**
  * A token cache that supports caching a token and refreshing it.
  */
 public class SimpleTokenCache {
-    private static final int REFRESH_TIMEOUT_SECONDS = 30;
-
-    private final AtomicBoolean wip;
-    private AccessToken cache;
-    private final ReplayProcessor<AccessToken> emitterProcessor = ReplayProcessor.create(1);
-    private final FluxSink<AccessToken> sink = emitterProcessor.sink(OverflowStrategy.BUFFER);
+    // The delay after a refresh to attempt another token refresh
+    private static final Duration REFRESH_DELAY = Duration.ofSeconds(30);
+    // the offset before token expiry to attempt proactive token refresh
+    private static final Duration REFRESH_OFFSET = Duration.ofMinutes(5);
+    private final AtomicReference<MonoProcessor<AccessToken>> wip;
+    private volatile AccessToken cache;
+    private volatile OffsetDateTime nextTokenRefresh = OffsetDateTime.now();
     private final Supplier<Mono<AccessToken>> tokenSupplier;
+    private final Predicate<AccessToken> shouldRefresh;
+    private final ClientLogger logger = new ClientLogger(SimpleTokenCache.class);
 
     /**
      * Creates an instance of RefreshableTokenCredential with default scheme "Bearer".
      *
      * @param tokenSupplier a method to get a new token
      */
     public SimpleTokenCache(Supplier<Mono<AccessToken>> tokenSupplier) {
-        this.wip = new AtomicBoolean(false);
+        this.wip = new AtomicReference<>();
         this.tokenSupplier = tokenSupplier;
+        this.shouldRefresh = accessToken -> OffsetDateTime.now()
+            .isAfter(accessToken.getExpiresAt().minus(REFRESH_OFFSET));
     }
 
     /**
      * Asynchronously get a token from either the cache or replenish the cache with a new token.
      * @return a Publisher that emits an AccessToken
      */
     public Mono<AccessToken> getToken() {
-        if (cache != null && !cache.isExpired()) {
-            return Mono.just(cache);
-        }
         return Mono.defer(() -> {
-            if (!wip.getAndSet(true)) {
-                return tokenSupplier.get().doOnNext(ac -> cache = ac)
-                    .doOnNext(sink::next)
-                    .doOnError(sink::error)
-                    .doOnTerminate(() -> wip.set(false));
-            } else {
-                return emitterProcessor.next();
+            try {
+                if (wip.compareAndSet(null, MonoProcessor.create())) {
+                    final MonoProcessor<AccessToken> monoProcessor = wip.get();
+                    OffsetDateTime now = OffsetDateTime.now();
+                    Mono<AccessToken> tokenRefresh;
+                    Mono<AccessToken> fallback;
+                    if (cache != null && !shouldRefresh.test(cache)) {
+                        // fresh cache & no need to refresh
+                        tokenRefresh = Mono.empty();
+                        fallback = Mono.just(cache);
+                    } else if (cache == null || cache.isExpired()) {
+                        // no token to use
+                        if (now.isAfter(nextTokenRefresh)) {
+                            // refresh immediately
+                            tokenRefresh = Mono.defer(tokenSupplier);
+                        } else {
+                            // wait for timeout, then refresh
+                            tokenRefresh = Mono.defer(tokenSupplier)
+                                .delaySubscription(Duration.between(now, nextTokenRefresh));
+                        }
+                        // cache doesn't exist or expired, no fallback
+                        fallback = Mono.empty();
+                    } else {
+                        // token available, but close to expiry
+                        if (now.isAfter(nextTokenRefresh)) {
+                            // refresh immediately
+                            tokenRefresh = Mono.defer(tokenSupplier);
+                        } else {
+                            // still in timeout, do not refresh
+                            tokenRefresh = Mono.empty();
+                        }
+                        // cache hasn't expired, ignore refresh error this time
+                        fallback = Mono.just(cache);
+                    }
+                    return tokenRefresh
+                        .materialize()
+                        .flatMap(signal -> {
+                            AccessToken accessToken = signal.get();
+                            Throwable error = signal.getThrowable();
+                            if (signal.isOnNext() && accessToken != null) { // SUCCESS
+                                logger.info(refreshLog(cache, now, "Acquired a new access token"));
+                                cache = accessToken;
+                                monoProcessor.onNext(accessToken);
+                                monoProcessor.onComplete();
+                                nextTokenRefresh = OffsetDateTime.now().plus(REFRESH_DELAY);
+                                return Mono.just(accessToken);
+                            } else if (signal.isOnError() && error != null) { // ERROR
+                                logger.error(refreshLog(cache, now, "Failed to acquire a new access token"));
+                                nextTokenRefresh = OffsetDateTime.now().plus(REFRESH_DELAY);
+                                return fallback.switchIfEmpty(Mono.error(error));
+                            } else { // NO REFRESH
+                                monoProcessor.onComplete();
+                                return fallback;
+                            }
+                        })
+                        .doOnError(monoProcessor::onError)
+                        .doOnTerminate(() -> wip.set(null));
+                } else if (cache != null && !cache.isExpired()) {
+                    // another thread might be refreshing the token proactively, but the current token is still valid
+                    return Mono.just(cache);
+                } else {
+                    // another thread is definitely refreshing the expired token
+                    MonoProcessor<AccessToken> monoProcessor = wip.get();
+                    if (monoProcessor == null) {
+                        // the refreshing thread has finished
+                        return Mono.just(cache);
+                    } else {
+                        // wait for refreshing thread to finish but defer to updated cache in case just missed onNext()
+                        return monoProcessor.switchIfEmpty(Mono.defer(() -> Mono.just(cache)));
+                    }
+                }
+            } catch (Throwable t) {
+                return Mono.error(t);
             }
         });
     }
+
+    private String refreshLog(AccessToken cache, OffsetDateTime now, String log) {
+        StringBuilder info = new StringBuilder(log);
+        if (cache == null) {
+            info.append(".");
+        } else {
+            Duration tte = Duration.between(now, cache.getExpiresAt());
+            info.append(" at ").append(tte.abs().getSeconds()).append(" seconds ")
+                .append(tte.isNegative() ? "after" : "before").append(" expiry. ")
+                .append("Retry may be attempted after ").append(REFRESH_DELAY.getSeconds()).append(" seconds.");
+            if (!tte.isNegative()) {
+                info.append(" The token currently cached will be used.");
+            }
+        }
+        return info.toString();
+    }
 }

sdk/e2e/pom.xml

@@ -33,7 +33,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-identity</artifactId>
-      <version>1.1.0-beta.7</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
+      <version>1.1.0-beta.8</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>

sdk/eventhubs/azure-messaging-eventhubs/pom.xml

@@ -37,7 +37,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.6.0</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      <version>1.7.0-beta.3</version> <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>
@@ -49,7 +49,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-identity</artifactId>
-      <version>1.0.9</version> <!-- {x-version-update;com.azure:azure-identity;dependency} -->
+      <version>1.1.0-beta.8</version> <!-- {x-version-update;unreleased_com.azure:azure-identity;dependency} -->
       <scope>test</scope>
     </dependency>
     <dependency>

sdk/identity/azure-identity-perf/pom.xml

@@ -27,7 +27,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-identity</artifactId>
-      <version>1.1.0-beta.7</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
+      <version>1.1.0-beta.8</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>

sdk/identity/azure-identity-perf/src/main/java/com/azure/identity/perf/WriteCache.java

@@ -9,16 +9,13 @@
 import com.azure.perf.test.core.PerfStressOptions;
 import reactor.core.publisher.Mono;
 
-import java.time.Duration;
-
 public class WriteCache extends ServiceTest<PerfStressOptions> {
     private final SharedTokenCacheCredential credential;
 
     public WriteCache(PerfStressOptions options) {
         super(options);
         credential = new SharedTokenCacheCredentialBuilder()
                 .clientId(CLI_CLIENT_ID)
-                .tokenRefreshOffset(Duration.ofMinutes(60))
                 .build();
     }
 

sdk/identity/azure-identity/CHANGELOG.md

@@ -1,8 +1,23 @@
 # Release History
 
-## 1.1.0-beta.7 (Unreleased)
-- Added support for web apps (confidential apps) for `InteractiveBrowserCredential` and `AuthorizationCodeCredential`. A client secret is required on the builder for web apps.
+## 1.1.0-beta.8 (Unreleased)
+
+
+## 1.1.0-beta.7 (2020-07-23)
+
+### Features
+- Added support for web apps (confidential apps) for `AuthorizationCodeCredential`. A client secret is required on the builder for web apps.
 - Added support for user assigned managed identities for `DefaultAzureCredential` with `.managedIdentityClientId()`.
+- Added`AzureAuthorityHosts` to access well knwon authority hosts.
+- Added `getClientId()` method in `AuthenticationRecord`
+
+### Breaking Changes
+- Removed persistent caching support from `AuthorizationCodeCredential`.
+- Removed `KnownAuthorityHosts`
+- Removed `getCredentials()` method in `ChainedTokenCredential` & `DefaultAzureCredential`
+- Changed return type of `serialize` method in `AuthenticationRecord` to `Mono<OutputStream>`.
+- Changed method signatures`enablePersistentCache(boolean)` and `allowUnencryptedCache(boolean)` on credential builders to `enablePersistentCache()` and `allowUnencryptedCache()`
+
 
 ## 1.1.0-beta.6 (2020-07-10)
 - Added `.getCredentials()` method to `DefaultAzureCredential` and `ChainedTokenCredential` and added option `.addAll(Collection<? extends TokenCredential>)` on `ChainedtokenCredentialBuilder`.

sdk/identity/azure-identity/pom.xml

@@ -6,7 +6,7 @@
 
   <groupId>com.azure</groupId>
   <artifactId>azure-identity</artifactId>
-  <version>1.1.0-beta.7</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
+  <version>1.1.0-beta.8</version> <!-- {x-version-update;com.azure:azure-identity;current} -->
 
   <name>Microsoft Azure client library for Identity</name>
   <description>This module contains client library for Microsoft Azure Identity.</description>
@@ -27,7 +27,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.6.0</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      <version>1.7.0-beta.3</version> <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.microsoft.azure</groupId>

sdk/identity/azure-identity/src/main/java/com/azure/identity/CredentialBuilderBase.java

@@ -89,21 +89,4 @@ public T httpClient(HttpClient client) {
         this.identityClientOptions.setHttpClient(client);
         return (T) this;
     }
-
-    /**
-     * Sets how long before the actual token expiry to refresh the token. The
-     * token will be considered expired at and after the time of (actual
-     * expiry - token refresh offset). The default offset is 2 minutes.
-     *
-     * This is useful when network is congested and a request containing the
-     * token takes longer than normal to get to the server.
-     *
-     * @param tokenRefreshOffset the duration before the actual expiry of a token to refresh it
-     * @return An updated instance of this builder with the token refresh offset set as specified.
-     */
-    @SuppressWarnings("unchecked")
-    public T tokenRefreshOffset(Duration tokenRefreshOffset) {
-        this.identityClientOptions.setTokenRefreshOffset(tokenRefreshOffset);
-        return (T) this;
-    }
 }

sdk/identity/azure-identity/src/main/java/com/azure/identity/DefaultAzureCredential.java

@@ -22,7 +22,6 @@
  */
 @Immutable
 public final class DefaultAzureCredential extends ChainedTokenCredential {
-
     /**
      * Creates default DefaultAzureCredential instance to use. This will use AZURE_CLIENT_ID,
      * AZURE_CLIENT_SECRET, and AZURE_TENANT_ID environment variables to create a

sdk/identity/azure-identity/src/main/java/com/azure/identity/EnvironmentCredential.java

@@ -37,7 +37,6 @@
 @Immutable
 public class EnvironmentCredential implements TokenCredential {
     private final Configuration configuration;
-    private final IdentityClientOptions identityClientOptions;
     private final ClientLogger logger = new ClientLogger(EnvironmentCredential.class);
     private final TokenCredential tokenCredential;
 
@@ -48,7 +47,6 @@ public class EnvironmentCredential implements TokenCredential {
      */
     EnvironmentCredential(IdentityClientOptions identityClientOptions) {
         this.configuration = Configuration.getGlobalConfiguration().clone();
-        this.identityClientOptions = identityClientOptions;
         TokenCredential targetCredential = null;
 
         String clientId = configuration.get(Configuration.PROPERTY_AZURE_CLIENT_ID);

sdk/identity/azure-identity/src/main/java/com/azure/identity/InteractiveBrowserCredential.java

@@ -122,5 +122,4 @@ private AccessToken updateCache(MsalToken msalToken) {
                                 identityClient.getTenantId(), identityClient.getClientId())));
         return msalToken;
     }
-
 }