Revert "Revert "Config subscription refactoring, part 5 [run-systemtest]""

2 files changed, 11 insertions, 21 deletions

diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java
index 5df7b1fc021..ab85c534251 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/RpcConfigSourceClient.java
@@ -4,7 +4,7 @@ package com.yahoo.vespa.config.proxy;
 import com.yahoo.concurrent.DaemonThreadFactory;
 import com.yahoo.config.ConfigurationRuntimeException;
 import com.yahoo.config.subscription.ConfigSourceSet;
-import com.yahoo.config.subscription.impl.JRTConfigRequester;
+import com.yahoo.config.subscription.impl.JrtConfigRequesters;
 import com.yahoo.jrt.Request;
 import com.yahoo.jrt.Spec;
 import com.yahoo.jrt.Supervisor;
@@ -51,7 +51,7 @@ class RpcConfigSourceClient implements ConfigSourceClient, Runnable {
     private final ScheduledExecutorService nextConfigScheduler =
             Executors.newScheduledThreadPool(1, new DaemonThreadFactory("next config"));
     private final ScheduledFuture<?> nextConfigFuture;
-    private final JRTConfigRequester requester;
+    private final JrtConfigRequesters requesters;
     // Scheduled executor that periodically checks for requests that have timed out and response should be returned to clients
     private final ScheduledExecutorService delayedResponsesScheduler =
             Executors.newScheduledThreadPool(1, new DaemonThreadFactory("delayed responses"));
@@ -64,7 +64,7 @@ class RpcConfigSourceClient implements ConfigSourceClient, Runnable {
         this.delayedResponses = new DelayedResponses();
         checkConfigSources();
         nextConfigFuture = nextConfigScheduler.scheduleAtFixedRate(this, 0, 10, MILLISECONDS);
-        this.requester = JRTConfigRequester.create(configSourceSet, timingValues);
+        this.requesters = new JrtConfigRequesters();
         DelayedResponseHandler command = new DelayedResponseHandler(delayedResponses, memoryCache, rpcServer);
         this.delayedResponsesFuture = delayedResponsesScheduler.scheduleAtFixedRate(command, 5, 1, SECONDS);
     }
@@ -142,7 +142,7 @@ class RpcConfigSourceClient implements ConfigSourceClient, Runnable {
         if (activeSubscribers.containsKey(configCacheKey)) return;
 
         log.log(Level.FINE, () -> "Could not find good config in cache, creating subscriber for: " + configCacheKey);
-        var subscriber = new Subscriber(input, configSourceSet, timingValues, requester);
+        var subscriber = new Subscriber(input, timingValues, requesters.getRequester(configSourceSet, timingValues));
         try {
             subscriber.subscribe();
             activeSubscribers.put(configCacheKey, subscriber);
@@ -187,12 +187,12 @@ class RpcConfigSourceClient implements ConfigSourceClient, Runnable {
         log.log(Level.FINE, "nextConfigScheduler.shutdownNow");
         nextConfigScheduler.shutdownNow();
         log.log(Level.FINE, "requester.close");
-        requester.close();
+        requesters.close();
     }
 
     @Override
     public String getActiveSourceConnection() {
-        return requester.getConnectionPool().getCurrent().getAddress();
+        return requesters.getRequester(configSourceSet, timingValues).getConnectionPool().getCurrent().getAddress();
     }
 
     @Override
diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/Subscriber.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/Subscriber.java
index 70ff4456f6c..b407c0e7e76 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/Subscriber.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/Subscriber.java
@@ -1,7 +1,6 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.config.proxy;
 
-import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.config.subscription.impl.GenericConfigHandle;
 import com.yahoo.config.subscription.impl.GenericConfigSubscriber;
 import com.yahoo.config.subscription.impl.JRTConfigRequester;
@@ -10,7 +9,6 @@ import com.yahoo.vespa.config.RawConfig;
 import com.yahoo.vespa.config.TimingValues;
 import com.yahoo.yolean.Exceptions;
 
-import java.util.Map;
 import java.util.Optional;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -23,22 +21,20 @@ public class Subscriber {
     private final static Logger log = Logger.getLogger(Subscriber.class.getName());
 
     private final RawConfig config;
-    private final ConfigSourceSet configSourceSet;
     private final TimingValues timingValues;
     private final GenericConfigSubscriber subscriber;
     private GenericConfigHandle handle;
 
-    Subscriber(RawConfig config, ConfigSourceSet configSourceSet, TimingValues timingValues, JRTConfigRequester requester) {
+    Subscriber(RawConfig config, TimingValues timingValues, JRTConfigRequester requester) {
         this.config = config;
-        this.configSourceSet = configSourceSet;
         this.timingValues = timingValues;
-        this.subscriber = new GenericConfigSubscriber(Map.of(configSourceSet, requester));
+        this.subscriber = new GenericConfigSubscriber(requester);
     }
 
     void subscribe() {
         ConfigKey<?> key = config.getKey();
         handle = subscriber.subscribe(new ConfigKey<>(key.getName(), key.getConfigId(), key.getNamespace()),
-                                      config.getDefContent(), configSourceSet, timingValues);
+                                      config.getDefContent(), timingValues);
     }
 
     public Optional<RawConfig> nextGeneration() {
@@ -58,14 +54,8 @@ public class Subscriber {
         return Optional.empty();
     }
 
-    public void cancel() {
-        if (subscriber != null) {
-            subscriber.close();
-        }
-    }
+    public void cancel() { subscriber.close(); }
 
-    boolean isClosed() {
-        return subscriber.isClosed();
-    }
+    boolean isClosed() { return subscriber.isClosed(); }
 
 }