Make DiscoveryServer stream ID global

When surfacing V2 and V3 streams alongside each other in envoy-control,
DiscoveryServerCallbacks are unable to differentiate between V2 & V3 ADS
upon onStreamClose(), onStreamCloseWithError().

This means that any DiscoveryServerCallback that keeps state cannot pivot
on stream IDs, since there will be duplicate between V2 & V3. This change
creates a global StreamCounter, which ensures stream IDs will be unique
across V2 & V3 streams.

Signed-off-by: Stephanie Tilden <[email protected]>

Author: Stephanie Tilden

server/src/main/java/io/envoyproxy/controlplane/server/DiscoveryServer.java

@@ -10,7 +10,6 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicLong;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -21,7 +20,6 @@ public abstract class DiscoveryServer<T, U> {
   final ConfigWatcher configWatcher;
   final ProtoResourcesSerializer protoResourcesSerializer;
   private final ExecutorGroup executorGroup;
-  private final AtomicLong streamCount = new AtomicLong();
 
   /**
    * Creates the server.
@@ -59,7 +57,7 @@ StreamObserver<T> createRequestHandler(
       boolean ads,
       String defaultTypeUrl) {
 
-    long streamId = streamCount.getAndIncrement();
+    long streamId = StreamCounter.getAndIncrement();
     Executor executor = executorGroup.next();
 
     LOGGER.debug("[{}] open stream from {}", streamId, defaultTypeUrl);

server/src/main/java/io/envoyproxy/controlplane/server/StreamCounter.java

@@ -0,0 +1,11 @@
+package io.envoyproxy.controlplane.server;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public final class StreamCounter {
+  private static final AtomicLong streamCount = new AtomicLong();
+
+  public static long getAndIncrement() {
+    return streamCount.getAndIncrement();
+  }
+}