Caching Refresh Token for Listeners

Previously, the refresh token caching mechanism for listeners used a
DirectProcessor.cache(1).  The had the effect of only caching a value if there
was already a subscriber attached.  As this should return a value on
subscription, regardless of the last time there was a subscriber, this change
updates the code to use a ReplayProcessor which guarentees the "last" value is
always available.

Author: nebhale

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

@@ -30,13 +30,11 @@
 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.core.publisher.ReplayProcessor;
 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;
@@ -75,7 +73,7 @@ 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, ReplayProcessor<String>> refreshTokenStreams = new ConcurrentHashMap<>(1);
 
     private final ConcurrentMap<ConnectionContext, Mono<String>> refreshTokens = new ConcurrentHashMap<>(1);
 
@@ -102,7 +100,7 @@ public String getClientSecret() {
      * @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();
+        return getRefreshTokenStream(connectionContext);
     }
 
     @Override
@@ -188,7 +186,7 @@ private Consumer<Map<String, String>> extractRefreshToken(ConnectionContext conn
                 }
 
                 this.refreshTokens.put(connectionContext, Mono.just(refreshToken));
-                getRefreshTokenStream(connectionContext).getT1().onNext(refreshToken);
+                getRefreshTokenStream(connectionContext).onNext(refreshToken);
             });
     }
 
@@ -201,11 +199,8 @@ 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 ReplayProcessor<String> getRefreshTokenStream(ConnectionContext connectionContext) {
+        return this.refreshTokenStreams.computeIfAbsent(connectionContext, c -> ReplayProcessor.create(1));
     }
 
     private Mono<HttpClientResponse> primaryToken(ConnectionContext connectionContext) {