xds: Use getWatchers more often in XdsDepManager

This provides better type and missing-map handling. Note that
getWatchers() now implicitly creates the map if it doesn't exist,
instead of just returning an empty map. That makes it a bit easier to
use and more importantly avoids accidents where a bug tries to modify
the immutable map.

Author: ejona86

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

@@ -112,14 +112,7 @@ private <T extends ResourceUpdate> void addWatcher(XdsWatcherBase<T> watcher) {
     XdsResourceType<T> type = watcher.type;
     String resourceName = watcher.resourceName;
 
-    @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);
+    getWatchers(type).put(resourceName, watcher);
     xdsClient.watchXdsResource(type, resourceName, watcher, syncContext);
   }
 
@@ -158,16 +151,12 @@ private <T extends ResourceUpdate> void cancelWatcher(XdsWatcherBase<T> watcher)
     XdsResourceType<T> type = watcher.type;
     String resourceName = watcher.resourceName;
 
-    @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);
+    if (getWatchers(type).remove(resourceName) == null) {
+      logger.log(DEBUG, "Trying to cancel watcher {0}, but it isn't watched", watcher);
       return;
     }
 
-    typeWatchers.watchers.remove(resourceName);
     xdsClient.cancelXdsResourceWatch(type, resourceName, watcher);
-
   }
 
   private static void throwIfParentContextsNotEmpty(XdsWatcherBase<?> watcher) {
@@ -213,8 +202,8 @@ private void releaseSubscription(ClusterSubscription subscription) {
         return;
       }
       subscription.closed = true;
-      XdsWatcherBase<?> cdsWatcher =
-          resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
+      XdsWatcherBase<XdsClusterResource.CdsUpdate> cdsWatcher
+          = getWatchers(CLUSTER_RESOURCE).get(clusterName);
       if (cdsWatcher == null) {
         return; // shutdown() called
       }
@@ -236,14 +225,12 @@ private void cancelClusterWatcherTree(CdsWatcher root, Object parentContext) {
     switch (cdsUpdate.clusterType()) {
       case EDS:
         String edsServiceName = root.getEdsServiceName();
-        EdsWatcher edsWatcher =
-            (EdsWatcher) resourceWatchers.get(ENDPOINT_RESOURCE).watchers.get(edsServiceName);
+        EdsWatcher edsWatcher = (EdsWatcher) getWatchers(ENDPOINT_RESOURCE).get(edsServiceName);
         cancelEdsWatcher(edsWatcher, root);
         break;
       case AGGREGATE:
         for (String cluster : cdsUpdate.prioritizedClusterNames()) {
-          CdsWatcher clusterWatcher =
-              (CdsWatcher) resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(cluster);
+          CdsWatcher clusterWatcher = (CdsWatcher) getWatchers(CLUSTER_RESOURCE).get(cluster);
           if (clusterWatcher != null) {
             cancelClusterWatcherTree(clusterWatcher, root);
           }
@@ -348,7 +335,8 @@ private <T extends ResourceUpdate> Map<String, XdsWatcherBase<T>> getWatchers(
       XdsResourceType<T> resourceType) {
     TypeWatchers<?> typeWatchers = resourceWatchers.get(resourceType);
     if (typeWatchers == null) {
-      return Collections.emptyMap();
+      typeWatchers = new TypeWatchers<T>(resourceType);
+      resourceWatchers.put(resourceType, typeWatchers);
     }
     assert typeWatchers.resourceType == resourceType;
     @SuppressWarnings("unchecked")
@@ -470,25 +458,22 @@ public String toString() {
 
   // Returns true if the watcher was added, false if it already exists
   private boolean addEdsWatcher(String edsServiceName, CdsWatcher parentContext) {
-    TypeWatchers<?> typeWatchers = resourceWatchers.get(XdsEndpointResource.getInstance());
-    if (typeWatchers == null || !typeWatchers.watchers.containsKey(edsServiceName)) {
-      addWatcher(new EdsWatcher(edsServiceName, parentContext));
-      return true;
+    EdsWatcher watcher
+        = (EdsWatcher) getWatchers(XdsEndpointResource.getInstance()).get(edsServiceName);
+    if (watcher != null) {
+      watcher.addParentContext(parentContext); // Is a set, so don't need to check for existence
+      return false;
     }
 
-    EdsWatcher watcher = (EdsWatcher) typeWatchers.watchers.get(edsServiceName);
-    watcher.addParentContext(parentContext); // Is a set, so don't need to check for existence
-    return false;
+    addWatcher(new EdsWatcher(edsServiceName, parentContext));
+    return true;
   }
 
   private void addClusterWatcher(String clusterName, Object parentContext, int depth) {
-    TypeWatchers<?> clusterWatchers = resourceWatchers.get(CLUSTER_RESOURCE);
-    if (clusterWatchers != null) {
-      CdsWatcher watcher = (CdsWatcher) clusterWatchers.watchers.get(clusterName);
-      if (watcher != null) {
-        watcher.parentContexts.put(parentContext, depth);
-        return;
-      }
+    CdsWatcher watcher = (CdsWatcher) getWatchers(CLUSTER_RESOURCE).get(clusterName);
+    if (watcher != null) {
+      watcher.parentContexts.put(parentContext, depth);
+      return;
     }
 
     addWatcher(new CdsWatcher(clusterName, parentContext, depth));
@@ -546,7 +531,7 @@ private static Set<String> getClusterNamesFromVirtualHost(VirtualHost virtualHos
   }
 
   private CdsWatcher getCluster(String clusterName) {
-    return (CdsWatcher) resourceWatchers.get(CLUSTER_RESOURCE).watchers.get(clusterName);
+    return (CdsWatcher) getWatchers(CLUSTER_RESOURCE).get(clusterName);
   }
 
   private static class TypeWatchers<T extends ResourceUpdate> {
@@ -557,10 +542,6 @@ private static class TypeWatchers<T extends ResourceUpdate> {
     TypeWatchers(XdsResourceType<T> resourceType) {
       this.resourceType = resourceType;
     }
-
-    public void add(String resourceName, XdsWatcherBase<T> watcher) {
-      watchers.put(resourceName, watcher);
-    }
   }
 
   public interface XdsConfigWatcher {
@@ -738,11 +719,11 @@ private void cleanUpRdsWatcher() {
         logger.log(XdsLogger.XdsLogLevel.DEBUG, "Stop watching RDS resource {0}", rdsName);
 
         // Cleanup clusters (as appropriate) that had the old rds watcher as a parent
-        if (!oldRdsWatcher.hasDataValue() || resourceWatchers.get(CLUSTER_RESOURCE) == null) {
+        if (!oldRdsWatcher.hasDataValue()) {
           return;
         }
-        for (XdsWatcherBase<?> watcher :
-            resourceWatchers.get(CLUSTER_RESOURCE).watchers.values()) {
+        for (XdsWatcherBase<XdsClusterResource.CdsUpdate> watcher :
+            getWatchers(CLUSTER_RESOURCE).values()) {
           cancelCdsWatcher((CdsWatcher) watcher, oldRdsWatcher);
         }
       }
@@ -752,11 +733,7 @@ private RdsWatcher getRdsWatcher() {
       if (rdsName == null) {
         return null;
       }
-      TypeWatchers<?> watchers = resourceWatchers.get(XdsRouteConfigureResource.getInstance());
-      if (watchers == null) {
-        return null;
-      }
-      return (RdsWatcher) watchers.watchers.get(rdsName);
+      return (RdsWatcher) getWatchers(XdsRouteConfigureResource.getInstance()).get(rdsName);
     }
 
     public RdsUpdateSupplier getRouteSource() {