xds: implement XdsClient APIs for watching LDS/RDS resources individually (#7470)

Add XdsClient implementation of watching LDS/RDS resources, replacing the ConfigWatcher API. This makes LDS/RDS/CDS/EDS resource watchers work similarly. This change also cleans up XdsClientImpl's tests.

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

@@ -16,6 +16,7 @@
 
 package io.grpc.xds;
 
+import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -117,7 +118,8 @@ abstract class XdsClient {
         @Nullable List<VirtualHost> virtualHosts) {
       this.httpMaxStreamDurationNano = httpMaxStreamDurationNano;
       this.rdsName = rdsName;
-      this.virtualHosts = virtualHosts;
+      this.virtualHosts = virtualHosts == null
+          ? null : Collections.unmodifiableList(new ArrayList<>(virtualHosts));
     }
 
     long getHttpMaxStreamDurationNano() {
@@ -169,7 +171,7 @@ abstract class XdsClient {
       return new Builder();
     }
 
-    private static class Builder {
+    static class Builder {
       private long httpMaxStreamDurationNano;
       @Nullable
       private String rdsName;
@@ -189,8 +191,11 @@ abstract class XdsClient {
         return this;
       }
 
-      Builder setVirtualHosts(List<VirtualHost> virtualHosts) {
-        this.virtualHosts = virtualHosts;
+      Builder addVirtualHost(VirtualHost virtualHost) {
+        if (virtualHosts == null) {
+          virtualHosts = new ArrayList<>();
+        }
+        virtualHosts.add(virtualHost);
         return this;
       }
 
@@ -206,7 +211,8 @@ abstract class XdsClient {
     private final List<VirtualHost> virtualHosts;
 
     private RdsUpdate(List<VirtualHost> virtualHosts) {
-      this.virtualHosts = virtualHosts;
+      this.virtualHosts = Collections.unmodifiableList(
+          new ArrayList<>(checkNotNull(virtualHosts, "virtualHosts")));
     }
 
     static RdsUpdate fromVirtualHosts(List<VirtualHost> virtualHosts) {
@@ -223,6 +229,23 @@ abstract class XdsClient {
           .add("virtualHosts", virtualHosts)
           .toString();
     }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(virtualHosts);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      RdsUpdate that = (RdsUpdate) o;
+      return Objects.equals(virtualHosts, that.virtualHosts);
+    }
   }
 
   static final class CdsUpdate implements ResourceUpdate {
@@ -474,7 +497,7 @@ abstract class XdsClient {
    * Updates via resource discovery RPCs using LDS. Includes {@link Listener} object containing
    * config for security, RBAC or other server side features such as rate limit.
    */
-  static final class ListenerUpdate {
+  static final class ListenerUpdate implements ResourceUpdate {
     // TODO(sanjaypujare): flatten structure by moving Listener class members here.
     private final Listener listener;
 

xds/src/test/java/io/grpc/xds/XdsClientTestHelper.java

@@ -56,6 +56,7 @@ import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
 import io.envoyproxy.envoy.type.v3.FractionalPercent;
 import io.envoyproxy.envoy.type.v3.FractionalPercent.DenominatorType;
 import io.grpc.xds.EnvoyProtoData.Node;
+import java.util.ArrayList;
 import java.util.List;
 import javax.annotation.Nullable;
 
@@ -160,6 +161,23 @@ class XdsClientTestHelper {
             .build();
   }
 
+  static List<VirtualHost> buildVirtualHosts(int num) {
+    List<VirtualHost> virtualHosts = new ArrayList<>(num);
+    for (int i = 0; i < num; i++) {
+      VirtualHost virtualHost =
+          VirtualHost.newBuilder()
+              .setName(num + ": do not care")
+              .addDomains("do not care")
+              .addRoutes(
+                  Route.newBuilder()
+                      .setRoute(RouteAction.newBuilder().setCluster("do not care"))
+                      .setMatch(RouteMatch.newBuilder().setPrefix("do not care")))
+              .build();
+      virtualHosts.add(virtualHost);
+    }
+    return virtualHosts;
+  }
+
   static VirtualHost buildVirtualHost(List<String> domains, String clusterName) {
     return VirtualHost.newBuilder()
         .setName("virtualhost00.googleapis.com") // don't care