Address a number of code review comments and add a test for missing C…

…ds aggregate and eds entries.

xds/src/main/java/io/grpc/xds/XdsConfig.java

@@ -18,6 +18,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.google.common.collect.ImmutableMap;
 import io.grpc.StatusOr;
 import io.grpc.xds.XdsClusterResource.CdsUpdate;
 import io.grpc.xds.XdsEndpointResource.EdsUpdate;
@@ -31,13 +32,18 @@
 /**
  * Represents the xDS configuration tree for a specified Listener.
  */
-public class XdsConfig {
-  final LdsUpdate listener;
-  final RdsUpdate route;
-  final Map<String, StatusOr<XdsClusterConfig>> clusters;
+final class XdsConfig {
+  private final LdsUpdate listener;
+  private final RdsUpdate route;
+  private final ImmutableMap<String, StatusOr<XdsClusterConfig>> clusters;
   private final int hashCode;
 
   XdsConfig(LdsUpdate listener, RdsUpdate route, Map<String, StatusOr<XdsClusterConfig>> clusters) {
+    this(listener, route, ImmutableMap.copyOf(clusters));
+  }
+
+  public XdsConfig(LdsUpdate listener, RdsUpdate route, ImmutableMap<String,
+      StatusOr<XdsClusterConfig>> clusters) {
     this.listener = listener;
     this.route = route;
     this.clusters = clusters;
@@ -53,8 +59,8 @@ public boolean equals(Object obj) {
 
     XdsConfig o = (XdsConfig) obj;
 
-    return Objects.equals(listener, o.listener) && Objects.equals(route, o.route)
-        && Objects.equals(clusters, o.clusters);
+    return hashCode() == o.hashCode() && Objects.equals(listener, o.listener)
+        && Objects.equals(route, o.route) && Objects.equals(clusters, o.clusters);
   }
 
   @Override
@@ -67,10 +73,22 @@ public String toString() {
     StringBuilder builder = new StringBuilder();
     builder.append("XdsConfig{listener=").append(listener)
         .append(", route=").append(route)
-        .append(", clusters={").append(clusters).append("}}");
+        .append(", clusters=").append(clusters).append("}");
     return builder.toString();
   }
 
+  public LdsUpdate getListener() {
+    return listener;
+  }
+
+  public RdsUpdate getRoute() {
+    return route;
+  }
+
+  public ImmutableMap<String, StatusOr<XdsClusterConfig>> getClusters() {
+    return clusters;
+  }
+
   public static class XdsClusterConfig {
     final String clusterName;
     final CdsUpdate clusterResource;
@@ -138,6 +156,8 @@ XdsConfigBuilder setRoute(RdsUpdate route) {
     }
 
     XdsConfigBuilder addCluster(String name, StatusOr<XdsClusterConfig> clusterConfig) {
+      checkNotNull(name, "name");
+      checkNotNull(clusterConfig, "clusterConfig");
       clusters.put(name, clusterConfig);
       return this;
     }

xds/src/main/java/io/grpc/xds/XdsDependencyManager.java

@@ -67,25 +67,27 @@ final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegi
                               String listenerName) {
     logId = InternalLogId.allocate("xds-dependency-manager", listenerName);
     logger = XdsLogger.withLogId(logId);
-    this.xdsClient = xdsClient;
-    this.xdsConfigWatcher = xdsConfigWatcher;
-    this.syncContext = syncContext;
+    this.xdsClient = checkNotNull(xdsClient, "xdsClient");
+    this.xdsConfigWatcher = checkNotNull(xdsConfigWatcher, "xdsConfigWatcher");
+    this.syncContext = checkNotNull(syncContext, "syncContext");
     this.dataPlaneAuthority = checkNotNull(dataPlaneAuthority, "dataPlaneAuthority");
 
     // start the ball rolling
-    addWatcher(new LdsWatcher(listenerName));
+    syncContext.execute(() -> addWatcher(new LdsWatcher(listenerName)));
   }
 
   @Override
-  public ClusterSubscription subscribeToCluster(String clusterName) {
+  public Closeable subscribeToCluster(String clusterName) {
 
     checkNotNull(clusterName, "clusterName");
     ClusterSubscription subscription = new ClusterSubscription(clusterName);
 
-    Set<ClusterSubscription> localSubscriptions =
-        clusterSubscriptions.computeIfAbsent(clusterName, k -> new HashSet<>());
-    localSubscriptions.add(subscription);
-    addWatcher(new CdsWatcher(clusterName));
+    syncContext.execute(() -> {
+      Set<ClusterSubscription> localSubscriptions =
+          clusterSubscriptions.computeIfAbsent(clusterName, k -> new HashSet<>());
+      localSubscriptions.add(subscription);
+      addWatcher(new CdsWatcher(clusterName));
+    });
 
     return subscription;
   }
@@ -95,50 +97,51 @@ private boolean hasWatcher(XdsResourceType<?> type, String resourceName) {
     return typeWatchers != null && typeWatchers.watchers.containsKey(resourceName);
   }
 
-  @SuppressWarnings("unchecked")
   private <T extends ResourceUpdate> void addWatcher(XdsWatcherBase<T> watcher) {
+    syncContext.throwIfNotInThisSynchronizationContext();
     XdsResourceType<T> type = watcher.type;
     String resourceName = watcher.resourceName;
 
-    this.syncContext.execute(() -> {
-      TypeWatchers<T> typeWatchers = (TypeWatchers<T>)resourceWatchers.get(type);
-      if (typeWatchers == null) {
-        typeWatchers = new TypeWatchers<>(type);
-        resourceWatchers.put(type, typeWatchers);
-      }
+    @SuppressWarnings("unchecked")
+    TypeWatchers<T> typeWatchers = (TypeWatchers<T>)resourceWatchers.get(type);
+    if (typeWatchers == null) {
+      typeWatchers = new TypeWatchers<>(type);
+      resourceWatchers.put(type, typeWatchers);
+    }
 
-      typeWatchers.add(resourceName, watcher);
-      xdsClient.watchXdsResource(type, resourceName, watcher);
-    });
+    typeWatchers.add(resourceName, watcher);
+    xdsClient.watchXdsResource(type, resourceName, watcher, syncContext);
   }
 
-  @SuppressWarnings("unchecked")
   private <T extends ResourceUpdate> void cancelWatcher(XdsWatcherBase<T> watcher) {
+    syncContext.throwIfNotInThisSynchronizationContext();
+
     if (watcher == null) {
       return;
     }
 
     XdsResourceType<T> type = watcher.type;
     String resourceName = watcher.resourceName;
 
-    this.syncContext.execute(() -> {
-      TypeWatchers<T> typeWatchers = (TypeWatchers<T>)resourceWatchers.get(type);
-      if (typeWatchers == null) {
-        logger.log(DEBUG, "Trying to cancel watcher {0}, but type not watched", watcher);
-        return;
-      }
+    @SuppressWarnings("unchecked")
+    TypeWatchers<T> typeWatchers = (TypeWatchers<T>)resourceWatchers.get(type);
+    if (typeWatchers == null) {
+      logger.log(DEBUG, "Trying to cancel watcher {0}, but type not watched", watcher);
+      return;
+    }
 
-      typeWatchers.watchers.remove(resourceName);
-      xdsClient.cancelXdsResourceWatch(type, resourceName, watcher);
-    });
+    typeWatchers.watchers.remove(resourceName);
+    xdsClient.cancelXdsResourceWatch(type, resourceName, watcher);
 
   }
 
   public void shutdown() {
-    for (TypeWatchers<?> watchers : resourceWatchers.values()) {
-      shutdownWatchersForType(watchers);
-    }
-    resourceWatchers.clear();
+    syncContext.execute(() -> {
+      for (TypeWatchers<?> watchers : resourceWatchers.values()) {
+        shutdownWatchersForType(watchers);
+      }
+      resourceWatchers.clear();
+    });
   }
 
   private <T extends ResourceUpdate> void shutdownWatchersForType(TypeWatchers<T> watchers) {
@@ -151,21 +154,21 @@ private <T extends ResourceUpdate> void shutdownWatchersForType(TypeWatchers<T>
   private void releaseSubscription(ClusterSubscription subscription) {
     checkNotNull(subscription, "subscription");
     String clusterName = subscription.getClusterName();
-    Set<ClusterSubscription> subscriptions = clusterSubscriptions.get(clusterName);
-    if (subscriptions == null) {
-      logger.log(DEBUG, "Subscription already released for {0}", clusterName);
-      return;
-    }
-
-    subscriptions.remove(subscription);
-    if (subscriptions.isEmpty()) {
-      clusterSubscriptions.remove(clusterName);
-      XdsWatcherBase<?> cdsWatcher =
-          resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
-      cancelClusterWatcherTree((CdsWatcher) cdsWatcher);
+    syncContext.execute(() -> {
+      Set<ClusterSubscription> subscriptions = clusterSubscriptions.get(clusterName);
+      if (subscriptions == null || !subscriptions.remove(subscription)) {
+        logger.log(DEBUG, "Subscription already released for {0}", clusterName);
+        return;
+      }
 
-      maybePublishConfig();
-    }
+      if (subscriptions.isEmpty()) {
+        clusterSubscriptions.remove(clusterName);
+        XdsWatcherBase<?> cdsWatcher =
+            resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
+        cancelClusterWatcherTree((CdsWatcher) cdsWatcher);
+        maybePublishConfig();
+      }
+    });
   }
 
   private void cancelClusterWatcherTree(CdsWatcher root) {
@@ -206,20 +209,18 @@ private void cancelClusterWatcherTree(CdsWatcher root) {
    * the watchers.
    */
   private void maybePublishConfig() {
-    syncContext.execute(() -> {
-      boolean waitingOnResource = resourceWatchers.values().stream()
-          .flatMap(typeWatchers -> typeWatchers.watchers.values().stream())
-          .anyMatch(watcher -> !watcher.hasResult());
-      if (waitingOnResource) {
-        return;
-      }
+    boolean waitingOnResource = resourceWatchers.values().stream()
+        .flatMap(typeWatchers -> typeWatchers.watchers.values().stream())
+        .anyMatch(watcher -> !watcher.hasResult());
+    if (waitingOnResource) {
+      return;
+    }
 
-      buildConfig();
-      xdsConfigWatcher.onUpdate(lastXdsConfig);
-    });
+    lastXdsConfig = buildConfig();
+    xdsConfigWatcher.onUpdate(lastXdsConfig);
   }
 
-  private void buildConfig() {
+  private XdsConfig buildConfig() {
     XdsConfig.XdsConfigBuilder builder = new XdsConfig.XdsConfigBuilder();
 
     // Iterate watchers and build the XdsConfig
@@ -258,7 +259,7 @@ private void buildConfig() {
       }
     }
 
-    lastXdsConfig = builder.build();
+    return builder.build();
   }
 
   @Override
@@ -334,7 +335,8 @@ public void onError(Status error) {
     protected void handleDoesNotExist(String resourceName) {
       checkArgument(this.resourceName.equals(resourceName), "Resource name does not match");
       data = StatusOr.fromStatus(
-          Status.UNAVAILABLE.withDescription("No " + type + " resource: " + resourceName));
+          Status.UNAVAILABLE
+              .withDescription("No " + type.typeName() + " resource: " + resourceName));
       transientError = false;
     }
 
@@ -362,7 +364,7 @@ boolean isTransientError() {
     }
 
     String toContextString() {
-      return type + " resource: " + resourceName;
+      return type.typeName() + " resource: " + resourceName;
     }
   }
 
@@ -503,9 +505,11 @@ public void onChanged(XdsClusterResource.CdsUpdate update) {
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
-      handleDoesNotExist(resourceName);
+      syncContext.execute(() -> {
+        handleDoesNotExist(resourceName);
+        maybePublishConfig();
+      });
     }
-
   }
 
   private class EdsWatcher extends XdsWatcherBase<XdsEndpointResource.EdsUpdate> {
@@ -523,7 +527,10 @@ public void onChanged(XdsEndpointResource.EdsUpdate update) {
 
     @Override
     public void onResourceDoesNotExist(String resourceName) {
-      handleDoesNotExist(resourceName);
+      syncContext.execute(() -> {
+        handleDoesNotExist(resourceName);
+        maybePublishConfig();
+      });
     }
   }
 

xds/src/main/java/io/grpc/xds/XdsRouteConfigureResource.java

@@ -136,7 +136,7 @@ protected RdsUpdate doParse(XdsResourceType.Args args, Message unpackedMessage)
         (RouteConfiguration) unpackedMessage, FilterRegistry.getDefaultRegistry(), args);
   }
 
-  static RdsUpdate processRouteConfiguration(
+  private static RdsUpdate processRouteConfiguration(
       RouteConfiguration routeConfig, FilterRegistry filterRegistry, XdsResourceType.Args args)
       throws ResourceInvalidException {
     return new RdsUpdate(extractVirtualHosts(routeConfig, filterRegistry, args));