server: Refactor stream observer

cache/src/main/java/io/envoyproxy/controlplane/cache/Watch.java

@@ -1,7 +1,6 @@
 package io.envoyproxy.controlplane.cache;
 
 import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
-
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.Consumer;
 

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

@@ -15,6 +15,10 @@
 import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 
+/**
+ * {@code AdsDiscoveryRequestStreamObserver} is an implementation of {@link DiscoveryRequestStreamObserver} tailored for
+ * ADS streams, which handle multiple watches for all TYPE_URLS.
+ */
 public class AdsDiscoveryRequestStreamObserver extends DiscoveryRequestStreamObserver {
   private final ConcurrentMap<String, Watch> watches;
   private final ConcurrentMap<String, DiscoveryResponse> latestResponse;

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

@@ -22,6 +22,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * {@code DiscoveryRequestStreamObserver} is provides the base implementation for XDS stream handling.
+ */
 public abstract class DiscoveryRequestStreamObserver implements StreamObserver<DiscoveryRequest> {
   private static final AtomicLongFieldUpdater<DiscoveryRequestStreamObserver> streamNonceUpdater =
       AtomicLongFieldUpdater.newUpdater(DiscoveryRequestStreamObserver.class, "streamNonce");

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

@@ -9,9 +9,14 @@
 import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 
+/**
+ * {@code XdsDiscoveryRequestStreamObserver} is a lightweight implementation of {@link DiscoveryRequestStreamObserver}
+ * tailored for non-ADS streams which handle a single watch.
+ */
 public class XdsDiscoveryRequestStreamObserver extends DiscoveryRequestStreamObserver {
   private volatile Watch watch;
   private volatile DiscoveryResponse latestResponse;
+  // ackedResources is only used in the same thread so it need not be volatile
   private Set<String> ackedResources;
 
   XdsDiscoveryRequestStreamObserver(String defaultTypeUrl,