Move single-thread Scheduler to ConnectionContext

Author: eaglerainbow

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

@@ -20,6 +20,7 @@
 import java.time.Duration;
 import java.util.Optional;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.netty.http.client.HttpClient;
 
 /**
@@ -52,6 +53,11 @@ public interface ConnectionContext {
      */
     RootProvider getRootProvider();
 
+    /**
+     * The {@link Scheduler} to use for token operations
+     */
+    Scheduler getTokenScheduler();
+
     /**
      * Attempt to explicitly trust the TLS certificate of an endpoint.  Implementations can choose whether any actual trusting will happen.
      *

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

@@ -32,6 +32,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.resources.ConnectionProvider;
 import reactor.netty.resources.LoopResources;
@@ -74,6 +76,7 @@ abstract class _DefaultConnectionContext implements ConnectionContext {
     @PreDestroy
     public final void dispose() {
         getConnectionProvider().ifPresent(ConnectionProvider::dispose);
+        getTokenScheduler().dispose();
         getThreadPool().dispose();
 
         try {
@@ -154,6 +157,12 @@ public Mono<Void> trust(String host, int port) {
             .orElse(Mono.empty());
     }
 
+    @Override
+    @Value.Derived
+    public Scheduler getTokenScheduler() {
+        return Schedulers.newSingle(String.format("token-provider-%s/%d", getApiHost(), getPort().orElse(DEFAULT_PORT)));
+    }
+
     /**
      * Additional configuration for the underlying HttpClient
      */

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

@@ -37,8 +37,6 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import reactor.core.scheduler.Scheduler;
-import reactor.core.scheduler.Schedulers;
 import org.cloudfoundry.Nullable;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
@@ -85,9 +83,6 @@ public abstract class AbstractUaaTokenProvider implements TokenProvider {
     private final ConcurrentMap<ConnectionContext, Mono<String>> refreshTokens =
             new ConcurrentHashMap<>(1);
 
-    private final ConcurrentMap<ConnectionContext, Scheduler> tokenSchedulers =
-            new ConcurrentHashMap<>(1);
-
     private final ConcurrentMap<ConnectionContext, Mono<String>> activeTokenRequests =
             new ConcurrentHashMap<>(1);
 
@@ -323,16 +318,10 @@ private Mono<String> token(final ConnectionContext connectionContext) {
          *   simply is to be considered waste.
          * 
          * The coding below fixes both issues: It ensures that the execution of the Mono
-         * is synchronized and it ensures that two threads arriving to fetch the JWT in a 
+         * is synchronized and it ensures that two threads arriving to fetch the JWT in a
          * non-caching situation does not trigger "wasteful" requests to the UAA server.
          */
 
-        // Get or create a single-threaded scheduler for this connection context
-        final Scheduler tokenScheduler = this.tokenSchedulers.computeIfAbsent(
-            connectionContext, 
-            ctx -> Schedulers.newSingle("token-" + ctx.hashCode())
-        );
-        
         /*
          * We use Mono.defer to ensure that the execution of the locking not happens
          * during creation of the Mono (where it is of little relevance), but
@@ -348,7 +337,7 @@ private Mono<String> token(final ConnectionContext connectionContext) {
 
             // Create new token request with proper synchronization
             final Mono<String> baseTokenRequest = createTokenRequest(connectionContext)
-                    .publishOn(tokenScheduler) // Ensure execution on single thread
+                    .publishOn(connectionContext.getTokenScheduler()) // Ensure execution on single thread
                     .doOnSubscribe(s -> LOGGER.debug("Starting new token request"))
                     .doOnSuccess(token -> LOGGER.debug("Token request completed successfully"))
                     .doOnError(error -> LOGGER.debug("Token request failed", error))