Merge branch '700-get-refresh-token'

Author: nebhale

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/TokenProvider.java

@@ -35,7 +35,7 @@ public interface TokenProvider {
      * Called when a {@code 401 UNAUTHORIZED} is received as part of a request.  Since not all {@link TokenProvider}s care about this possibility, the default implementation does nothing.
      * Implementations are free to manage internal state with this call if they choose to.
      *
-     * @param connectionContext A {@link ConnectionContext} to be used if a token needs to be retrieved via a network request
+     * @param connectionContext A {@link ConnectionContext} to be used to identity which connection the tokens should be invalidated for
      */
     default void invalidate(ConnectionContext connectionContext) {
     }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/tokenprovider/AbstractUaaTokenProvider.java

@@ -30,9 +30,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.web.util.UriComponentsBuilder;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.ipc.netty.http.client.HttpClientRequest;
 import reactor.ipc.netty.http.client.HttpClientResponse;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 import java.time.LocalDateTime;
 import java.time.ZoneId;
@@ -71,6 +75,8 @@ public abstract class AbstractUaaTokenProvider implements TokenProvider {
 
     private final ConcurrentMap<ConnectionContext, Mono<String>> accessTokens = new ConcurrentHashMap<>(1);
 
+    private final ConcurrentMap<ConnectionContext, Tuple2<DirectProcessor<String>, Flux<String>>> refreshTokenStreams = new ConcurrentHashMap<>(1);
+
     private final ConcurrentMap<ConnectionContext, Mono<String>> refreshTokens = new ConcurrentHashMap<>(1);
 
     /**
@@ -89,6 +95,16 @@ public String getClientSecret() {
         return "";
     }
 
+    /**
+     * Returns a {@link Flux} of refresh tokens for a connection
+     *
+     * @param connectionContext A {@link ConnectionContext} to be used to identity which connection the refresh tokens be retrieved for
+     * @return a {@link Flux} that emits the last token on subscribe and new refresh tokens as they are negotiated
+     */
+    public Flux<String> getRefreshTokens(ConnectionContext connectionContext) {
+        return getRefreshTokenStream(connectionContext).getT2();
+    }
+
     @Override
     public final Mono<String> getToken(ConnectionContext connectionContext) {
         return this.accessTokens.computeIfAbsent(connectionContext, this::token);
@@ -172,6 +188,7 @@ private Consumer<Map<String, String>> extractRefreshToken(ConnectionContext conn
                 }
 
                 this.refreshTokens.put(connectionContext, Mono.just(refreshToken));
+                getRefreshTokenStream(connectionContext).getT1().onNext(refreshToken);
             });
     }
 
@@ -184,6 +201,13 @@ private Function<Mono<HttpClientResponse>, Mono<String>> extractTokens(Connectio
             .map(AbstractUaaTokenProvider::extractAccessToken);
     }
 
+    private Tuple2<DirectProcessor<String>, Flux<String>> getRefreshTokenStream(ConnectionContext connectionContext) {
+        return this.refreshTokenStreams.computeIfAbsent(connectionContext, c -> {
+            DirectProcessor<String> processor = DirectProcessor.create();
+            return Tuples.of(processor, processor.cache(1));
+        });
+    }
+
     private Mono<HttpClientResponse> primaryToken(ConnectionContext connectionContext) {
         return requestToken(connectionContext, this::tokenRequestTransformer);
     }