Second attempt: XdsClient: remove resource-type-specific methods from XdsClient API (#28305)

* Revert "Revert "XdsClient: remove resource-type-specific methods from XdsClient API (#28231)" (#28301)"

This reverts commit 3e779b68fe.

* remove global resource type registry

src/core/ext/filters/client_channel/lb_policy/xds/cds.cc

@@ -25,6 +25,7 @@
 #include "src/core/ext/filters/client_channel/lb_policy_registry.h"
 #include "src/core/ext/xds/xds_certificate_provider.h"
 #include "src/core/ext/xds/xds_client.h"
+#include "src/core/ext/xds/xds_cluster.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/gprpp/memory.h"
 #include "src/core/lib/gprpp/orphanable.h"
@@ -66,7 +67,7 @@ class CdsLb : public LoadBalancingPolicy {
 
  private:
   // Watcher for getting cluster data from XdsClient.
-  class ClusterWatcher : public XdsClient::ClusterWatcherInterface {
+  class ClusterWatcher : public XdsClusterResourceType::WatcherInterface {
    public:
     ClusterWatcher(RefCountedPtr<CdsLb> parent, std::string name)
         : parent_(std::move(parent)), name_(std::move(name)) {}
@@ -147,7 +148,7 @@ class CdsLb : public LoadBalancingPolicy {
       const std::string& cluster_name, const XdsClusterResource& cluster_data);
 
   void CancelClusterDataWatch(absl::string_view cluster_name,
-                              XdsClient::ClusterWatcherInterface* watcher,
+                              ClusterWatcher* watcher,
                               bool delay_unsubscription = false);
 
   void MaybeDestroyChildPolicyLocked();
@@ -301,7 +302,8 @@ void CdsLb::UpdateLocked(UpdateArgs args) {
     }
     auto watcher = MakeRefCounted<ClusterWatcher>(Ref(), config_->cluster());
     watchers_[config_->cluster()].watcher = watcher.get();
-    xds_client_->WatchClusterData(config_->cluster(), std::move(watcher));
+    XdsClusterResourceType::StartWatch(xds_client_.get(), config_->cluster(),
+                                       std::move(watcher));
   }
 }
 
@@ -326,7 +328,8 @@ bool CdsLb::GenerateDiscoveryMechanismForCluster(
               name.c_str());
     }
     state.watcher = watcher.get();
-    xds_client_->WatchClusterData(name, std::move(watcher));
+    XdsClusterResourceType::StartWatch(xds_client_.get(), name,
+                                       std::move(watcher));
     return false;
   }
   // Don't have the update we need yet.
@@ -610,7 +613,7 @@ grpc_error_handle CdsLb::UpdateXdsCertificateProvider(
 }
 
 void CdsLb::CancelClusterDataWatch(absl::string_view cluster_name,
-                                   XdsClient::ClusterWatcherInterface* watcher,
+                                   ClusterWatcher* watcher,
                                    bool delay_unsubscription) {
   if (xds_certificate_provider_ != nullptr) {
     std::string name(cluster_name);
@@ -620,7 +623,7 @@ void CdsLb::CancelClusterDataWatch(absl::string_view cluster_name,
                                                                     nullptr);
     xds_certificate_provider_->UpdateSubjectAlternativeNameMatchers(name, {});
   }
-  xds_client_->CancelClusterDataWatch(cluster_name, watcher,
+  XdsClusterResourceType::CancelWatch(xds_client_.get(), cluster_name, watcher,
                                       delay_unsubscription);
 }
 //

src/core/ext/filters/client_channel/lb_policy/xds/xds_cluster_impl.cc

@@ -30,6 +30,7 @@
 #include "src/core/ext/filters/client_channel/lb_policy_registry.h"
 #include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_client_stats.h"
+#include "src/core/ext/xds/xds_endpoint.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/gpr/env.h"
 #include "src/core/lib/gpr/string.h"

src/core/ext/filters/client_channel/lb_policy/xds/xds_cluster_resolver.cc

@@ -39,6 +39,7 @@
 #include "src/core/ext/xds/xds_channel_args.h"
 #include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_client_stats.h"
+#include "src/core/ext/xds/xds_endpoint.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/gpr/string.h"
 #include "src/core/lib/gprpp/orphanable.h"
@@ -165,7 +166,7 @@ class XdsClusterResolverLb : public LoadBalancingPolicy {
     bool disable_reresolution() override { return true; }
 
    private:
-    class EndpointWatcher : public XdsClient::EndpointWatcherInterface {
+    class EndpointWatcher : public XdsEndpointResourceType::WatcherInterface {
      public:
       explicit EndpointWatcher(
           RefCountedPtr<EdsDiscoveryMechanism> discovery_mechanism)
@@ -414,8 +415,8 @@ void XdsClusterResolverLb::EdsDiscoveryMechanism::Start() {
   auto watcher = MakeRefCounted<EndpointWatcher>(
       Ref(DEBUG_LOCATION, "EdsDiscoveryMechanism"));
   watcher_ = watcher.get();
-  parent()->xds_client_->WatchEndpointData(GetEdsResourceName(),
+  XdsEndpointResourceType::StartWatch(parent()->xds_client_.get(),
-                                           std::move(watcher));
+                                      GetEdsResourceName(), std::move(watcher));
 }
 
 void XdsClusterResolverLb::EdsDiscoveryMechanism::Orphan() {
@@ -425,8 +426,8 @@ void XdsClusterResolverLb::EdsDiscoveryMechanism::Orphan() {
             ":%p cancelling xds watch for %s",
             parent(), index(), this, std::string(GetEdsResourceName()).c_str());
   }
-  parent()->xds_client_->CancelEndpointDataWatch(GetEdsResourceName(),
+  XdsEndpointResourceType::CancelWatch(parent()->xds_client_.get(),
-                                                 watcher_);
+                                       GetEdsResourceName(), watcher_);
   Unref();
 }
 

src/core/ext/filters/client_channel/resolver/xds/xds_resolver.cc

@@ -31,6 +31,8 @@
 #include "src/core/ext/xds/xds_channel_args.h"
 #include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_http_filters.h"
+#include "src/core/ext/xds/xds_listener.h"
+#include "src/core/ext/xds/xds_route_config.h"
 #include "src/core/ext/xds/xds_routing.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/closure.h"
@@ -80,7 +82,7 @@ class XdsResolver : public Resolver {
   }
 
  private:
-  class ListenerWatcher : public XdsClient::ListenerWatcherInterface {
+  class ListenerWatcher : public XdsListenerResourceType::WatcherInterface {
    public:
     explicit ListenerWatcher(RefCountedPtr<XdsResolver> resolver)
         : resolver_(std::move(resolver)) {}
@@ -118,7 +120,8 @@ class XdsResolver : public Resolver {
     RefCountedPtr<XdsResolver> resolver_;
   };
 
-  class RouteConfigWatcher : public XdsClient::RouteConfigWatcherInterface {
+  class RouteConfigWatcher
+      : public XdsRouteConfigResourceType::WatcherInterface {
    public:
     explicit RouteConfigWatcher(RefCountedPtr<XdsResolver> resolver)
         : resolver_(std::move(resolver)) {}
@@ -291,14 +294,14 @@ class XdsResolver : public Resolver {
 
   RefCountedPtr<XdsClient> xds_client_;
 
-  XdsClient::ListenerWatcherInterface* listener_watcher_ = nullptr;
+  ListenerWatcher* listener_watcher_ = nullptr;
   // This will not contain the RouteConfiguration, even if it comes with the
   // LDS response; instead, the relevant VirtualHost from the
   // RouteConfiguration will be saved in current_virtual_host_.
   XdsListenerResource current_listener_;
 
   std::string route_config_name_;
-  XdsClient::RouteConfigWatcherInterface* route_config_watcher_ = nullptr;
+  RouteConfigWatcher* route_config_watcher_ = nullptr;
   XdsRouteConfigResource::VirtualHost current_virtual_host_;
 
   ClusterState::ClusterStateMap cluster_state_map_;
@@ -694,7 +697,8 @@ void XdsResolver::StartLocked() {
                                    interested_parties_);
   auto watcher = MakeRefCounted<ListenerWatcher>(Ref());
   listener_watcher_ = watcher.get();
-  xds_client_->WatchListenerData(server_name_, std::move(watcher));
+  XdsListenerResourceType::StartWatch(xds_client_.get(), server_name_,
+                                      std::move(watcher));
 }
 
 void XdsResolver::ShutdownLocked() {
@@ -703,12 +707,14 @@ void XdsResolver::ShutdownLocked() {
   }
   if (xds_client_ != nullptr) {
     if (listener_watcher_ != nullptr) {
-      xds_client_->CancelListenerDataWatch(server_name_, listener_watcher_,
+      XdsListenerResourceType::CancelWatch(xds_client_.get(), server_name_,
+                                           listener_watcher_,
                                            /*delay_unsubscription=*/false);
     }
     if (route_config_watcher_ != nullptr) {
-      xds_client_->CancelRouteConfigDataWatch(
+      XdsRouteConfigResourceType::CancelWatch(
-          server_name_, route_config_watcher_, /*delay_unsubscription=*/false);
+          xds_client_.get(), route_config_name_, route_config_watcher_,
+          /*delay_unsubscription=*/false);
     }
     grpc_pollset_set_del_pollset_set(xds_client_->interested_parties(),
                                      interested_parties_);
@@ -726,8 +732,8 @@ void XdsResolver::OnListenerUpdate(XdsListenerResource listener) {
   if (listener.http_connection_manager.route_config_name !=
       route_config_name_) {
     if (route_config_watcher_ != nullptr) {
-      xds_client_->CancelRouteConfigDataWatch(
+      XdsRouteConfigResourceType::CancelWatch(
-          route_config_name_, route_config_watcher_,
+          xds_client_.get(), route_config_name_, route_config_watcher_,
           /*delay_unsubscription=*/
           !listener.http_connection_manager.route_config_name.empty());
       route_config_watcher_ = nullptr;
@@ -738,7 +744,8 @@ void XdsResolver::OnListenerUpdate(XdsListenerResource listener) {
       current_virtual_host_.routes.clear();
       auto watcher = MakeRefCounted<RouteConfigWatcher>(Ref());
       route_config_watcher_ = watcher.get();
-      xds_client_->WatchRouteConfigData(route_config_name_, std::move(watcher));
+      XdsRouteConfigResourceType::StartWatch(
+          xds_client_.get(), route_config_name_, std::move(watcher));
     }
   }
   current_listener_ = std::move(listener);

src/core/ext/xds/xds_api.cc

@@ -82,11 +82,13 @@ namespace grpc_core {
 XdsApi::XdsApi(XdsClient* client, TraceFlag* tracer,
                const XdsBootstrap::Node* node,
                const CertificateProviderStore::PluginDefinitionMap*
-                   certificate_provider_definition_map)
+                   certificate_provider_definition_map,
+               upb::SymbolTable* symtab)
     : client_(client),
       tracer_(tracer),
       node_(node),
       certificate_provider_definition_map_(certificate_provider_definition_map),
+      symtab_(symtab),
       build_version_(absl::StrCat("gRPC C-core ", GPR_PLATFORM_STRING, " ",
                                   grpc_version_string(),
                                   GRPC_XDS_USER_AGENT_NAME_SUFFIX_STRING,
@@ -96,16 +98,7 @@ XdsApi::XdsApi(XdsClient* client, TraceFlag* tracer,
       user_agent_version_(
           absl::StrCat("C-core ", grpc_version_string(),
                        GRPC_XDS_USER_AGENT_NAME_SUFFIX_STRING,
-                       GRPC_XDS_USER_AGENT_VERSION_SUFFIX_STRING)) {
+                       GRPC_XDS_USER_AGENT_VERSION_SUFFIX_STRING)) {}
-  // Populate upb symtab with xDS proto messages that we want to print
-  // properly in logs.
-  // Note: This won't actually work properly until upb adds support for
-  // Any fields in textproto printing (internal b/178821188).
-  XdsResourceTypeRegistry::GetOrCreate()->ForEach(
-      [this](const XdsResourceType* type) {
-        type->InitUpbSymtab(symtab_.ptr());
-      });
-}
 
 namespace {
 
@@ -284,7 +277,7 @@ grpc_slice XdsApi::CreateAdsRequest(
   upb::Arena arena;
   const XdsEncodingContext context = {client_,
                                       tracer_,
-                                      symtab_.ptr(),
+                                      symtab_->ptr(),
                                       arena.ptr(),
                                       server.ShouldUseV3(),
                                       certificate_provider_definition_map_};
@@ -363,7 +356,7 @@ absl::Status XdsApi::ParseAdsResponse(const XdsBootstrap::XdsServer& server,
   upb::Arena arena;
   const XdsEncodingContext context = {client_,
                                       tracer_,
-                                      symtab_.ptr(),
+                                      symtab_->ptr(),
                                       arena.ptr(),
                                       server.ShouldUseV3(),
                                       certificate_provider_definition_map_};
@@ -438,7 +431,7 @@ grpc_slice XdsApi::CreateLrsInitialRequest(
   upb::Arena arena;
   const XdsEncodingContext context = {client_,
                                       tracer_,
-                                      symtab_.ptr(),
+                                      symtab_->ptr(),
                                       arena.ptr(),
                                       server.ShouldUseV3(),
                                       certificate_provider_definition_map_};
@@ -512,7 +505,7 @@ grpc_slice XdsApi::CreateLrsRequest(
     ClusterLoadReportMap cluster_load_report_map) {
   upb::Arena arena;
   const XdsEncodingContext context = {
-      client_,     tracer_, symtab_.ptr(),
+      client_,     tracer_, symtab_->ptr(),
       arena.ptr(), false,   certificate_provider_definition_map_};
   // Create a request.
   envoy_service_load_stats_v3_LoadStatsRequest* request =
@@ -636,7 +629,7 @@ std::string XdsApi::AssembleClientConfig(
   auto* node = envoy_service_status_v3_ClientConfig_mutable_node(client_config,
                                                                  arena.ptr());
   const XdsEncodingContext context = {
-      client_,     tracer_, symtab_.ptr(),
+      client_,     tracer_, symtab_->ptr(),
       arena.ptr(), true,    certificate_provider_definition_map_};
   PopulateNode(context, node_, build_version_, user_agent_name_,
                user_agent_version_, node);

src/core/ext/xds/xds_api.h

@@ -134,7 +134,8 @@ class XdsApi {
                 "");
 
   XdsApi(XdsClient* client, TraceFlag* tracer, const XdsBootstrap::Node* node,
-         const CertificateProviderStore::PluginDefinitionMap* map);
+         const CertificateProviderStore::PluginDefinitionMap* map,
+         upb::SymbolTable* symtab);
 
   // Creates an ADS request.
   // Takes ownership of \a error.
@@ -175,7 +176,7 @@ class XdsApi {
   const XdsBootstrap::Node* node_;  // Do not own.
   const CertificateProviderStore::PluginDefinitionMap*
       certificate_provider_definition_map_;  // Do not own.
-  upb::SymbolTable symtab_;
+  upb::SymbolTable* symtab_;                 // Do not own.
   const std::string build_version_;
   const std::string user_agent_name_;
   const std::string user_agent_version_;

src/core/ext/xds/xds_client.cc

@@ -37,7 +37,10 @@
 #include "src/core/ext/xds/xds_bootstrap.h"
 #include "src/core/ext/xds/xds_channel_args.h"
 #include "src/core/ext/xds/xds_client_stats.h"
+#include "src/core/ext/xds/xds_cluster.h"
+#include "src/core/ext/xds/xds_endpoint.h"
 #include "src/core/ext/xds/xds_http_filters.h"
+#include "src/core/ext/xds/xds_listener.h"
 #include "src/core/lib/address_utils/sockaddr_utils.h"
 #include "src/core/lib/backoff/backoff.h"
 #include "src/core/lib/channel/channel_args.h"
@@ -76,11 +79,6 @@ const grpc_channel_args* g_channel_args ABSL_GUARDED_BY(*g_mu) = nullptr;
 XdsClient* g_xds_client ABSL_GUARDED_BY(*g_mu) = nullptr;
 char* g_fallback_bootstrap_config ABSL_GUARDED_BY(*g_mu) = nullptr;
 
-const char* kLdsTypeUrl = "envoy.config.listener.v3.Listener";
-const char* kRdsTypeUrl = "envoy.config.route.v3.RouteConfiguration";
-const char* kCdsTypeUrl = "envoy.config.cluster.v3.Cluster";
-const char* kEdsTypeUrl = "envoy.config.endpoint.v3.ClusterLoadAssignment";
-
 }  // namespace
 
 class XdsClient::Notifier {
@@ -202,7 +200,8 @@ class XdsClient::ChannelState::AdsCallState
     explicit AdsResponseParser(AdsCallState* ads_call_state)
         : ads_call_state_(ads_call_state) {}
 
-    absl::Status ProcessAdsResponseFields(AdsResponseFields fields) override;
+    absl::Status ProcessAdsResponseFields(AdsResponseFields fields) override
+        ABSL_EXCLUSIVE_LOCKS_REQUIRED(&XdsClient::mu_);
 
     void ParseResource(const XdsEncodingContext& context, size_t idx,
                        absl::string_view type_url,
@@ -745,7 +744,7 @@ absl::Status XdsClient::ChannelState::AdsCallState::AdsResponseParser::
             fields.version.c_str(), fields.nonce.c_str(), fields.num_resources);
   }
   result_.type =
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(fields.type_url);
+      ads_call_state_->xds_client()->GetResourceTypeLocked(fields.type_url);
   if (result_.type == nullptr) {
     return absl::InvalidArgumentError(
         absl::StrCat("unknown resource type ", fields.type_url));
@@ -1778,7 +1777,7 @@ XdsClient::XdsClient(std::unique_ptr<XdsBootstrap> bootstrap,
       certificate_provider_store_(MakeOrphanable<CertificateProviderStore>(
           bootstrap_->certificate_providers())),
       api_(this, &grpc_xds_client_trace, bootstrap_->node(),
-           &bootstrap_->certificate_providers()) {
+           &bootstrap_->certificate_providers(), &symtab_) {
   if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
     gpr_log(GPR_INFO, "[xds_client %p] creating xds client", this);
   }
@@ -1836,6 +1835,7 @@ void XdsClient::WatchResource(const XdsResourceType* type,
   if (!resource_name.ok()) {
     {
       MutexLock lock(&mu_);
+      MaybeRegisterResourceTypeLocked(type);
       invalid_watchers_[w] = watcher;
     }
     grpc_error_handle error = GRPC_ERROR_CREATE_FROM_CPP_STRING(
@@ -1851,6 +1851,7 @@ void XdsClient::WatchResource(const XdsResourceType* type,
   }
   {
     MutexLock lock(&mu_);
+    MaybeRegisterResourceTypeLocked(type);
     // TODO(donnadionne): If we get a request for an authority that is not
     // configured in the bootstrap file, reject it.
     AuthorityState& authority_state =
@@ -1924,64 +1925,25 @@ void XdsClient::CancelResourceWatch(const XdsResourceType* type,
   }
 }
 
-void XdsClient::WatchListenerData(
+void XdsClient::MaybeRegisterResourceTypeLocked(
-    absl::string_view listener_name,
+    const XdsResourceType* resource_type) {
-    RefCountedPtr<ListenerWatcherInterface> watcher) {
+  auto it = resource_types_.find(resource_type->type_url());
-  WatchResource(XdsResourceTypeRegistry::GetOrCreate()->GetType(kLdsTypeUrl),
+  if (it != resource_types_.end()) {
-                listener_name, std::move(watcher));
+    GPR_ASSERT(it->second == resource_type);
-}
+    return;
-
+  }
-void XdsClient::CancelListenerDataWatch(absl::string_view listener_name,
+  resource_types_.emplace(resource_type->type_url(), resource_type);
-                                        ListenerWatcherInterface* watcher,
+  v2_resource_types_.emplace(resource_type->v2_type_url(), resource_type);
-                                        bool delay_unsubscription) {
+  resource_type->InitUpbSymtab(symtab_.ptr());
-  CancelResourceWatch(
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(kLdsTypeUrl),
-      listener_name, watcher, delay_unsubscription);
-}
-
-void XdsClient::WatchRouteConfigData(
-    absl::string_view route_config_name,
-    RefCountedPtr<RouteConfigWatcherInterface> watcher) {
-  WatchResource(XdsResourceTypeRegistry::GetOrCreate()->GetType(kRdsTypeUrl),
-                route_config_name, std::move(watcher));
-}
-
-void XdsClient::CancelRouteConfigDataWatch(absl::string_view route_config_name,
-                                           RouteConfigWatcherInterface* watcher,
-                                           bool delay_unsubscription) {
-  CancelResourceWatch(
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(kRdsTypeUrl),
-      route_config_name, watcher, delay_unsubscription);
-}
-
-void XdsClient::WatchClusterData(
-    absl::string_view cluster_name,
-    RefCountedPtr<ClusterWatcherInterface> watcher) {
-  WatchResource(XdsResourceTypeRegistry::GetOrCreate()->GetType(kCdsTypeUrl),
-                cluster_name, std::move(watcher));
-}
-
-void XdsClient::CancelClusterDataWatch(absl::string_view cluster_name,
-                                       ClusterWatcherInterface* watcher,
-                                       bool delay_unsubscription) {
-  CancelResourceWatch(
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(kCdsTypeUrl),
-      cluster_name, watcher, delay_unsubscription);
-}
-
-void XdsClient::WatchEndpointData(
-    absl::string_view eds_service_name,
-    RefCountedPtr<EndpointWatcherInterface> watcher) {
-  WatchResource(XdsResourceTypeRegistry::GetOrCreate()->GetType(kEdsTypeUrl),
-                eds_service_name, std::move(watcher));
 }
 
-void XdsClient::CancelEndpointDataWatch(absl::string_view eds_service_name,
+const XdsResourceType* XdsClient::GetResourceTypeLocked(
-                                        EndpointWatcherInterface* watcher,
+    absl::string_view resource_type) {
-                                        bool delay_unsubscription) {
+  auto it = resource_types_.find(resource_type);
-  CancelResourceWatch(
+  if (it != resource_types_.end()) return it->second;
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(kEdsTypeUrl),
+  auto it2 = v2_resource_types_.find(resource_type);
-      eds_service_name, watcher, delay_unsubscription);
+  if (it2 != v2_resource_types_.end()) return it2->second;
+  return nullptr;
 }
 
 absl::StatusOr<XdsClient::XdsResourceName> XdsClient::ParseXdsResourceName(
@@ -2050,9 +2012,8 @@ RefCountedPtr<XdsClusterDropStats> XdsClient::AddClusterDropStats(
         it->first.second /*eds_service_name*/);
     load_report_state.drop_stats = cluster_drop_stats.get();
   }
-  auto resource_name = ParseXdsResourceName(
+  auto resource_name =
-      cluster_name,
+      ParseXdsResourceName(cluster_name, XdsClusterResourceType::Get());
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(kCdsTypeUrl));
   GPR_ASSERT(resource_name.ok());
   auto a = authority_state_map_.find(resource_name->authority);
   if (a != authority_state_map_.end()) {
@@ -2114,9 +2075,8 @@ RefCountedPtr<XdsClusterLocalityStats> XdsClient::AddClusterLocalityStats(
         std::move(locality));
     locality_state.locality_stats = cluster_locality_stats.get();
   }
-  auto resource_name = ParseXdsResourceName(
+  auto resource_name =
-      cluster_name,
+      ParseXdsResourceName(cluster_name, XdsClusterResourceType::Get());
-      XdsResourceTypeRegistry::GetOrCreate()->GetType(kCdsTypeUrl));
   GPR_ASSERT(resource_name.ok());
   auto a = authority_state_map_.find(resource_name->authority);
   if (a != authority_state_map_.end()) {
@@ -2285,23 +2245,9 @@ std::string XdsClient::DumpClientConfigBinary() {
 // accessors for global state
 //
 
-namespace {
-
-void InitResourceTypeRegistry() {
-  auto* registry = XdsResourceTypeRegistry::GetOrCreate();
-  registry->RegisterType(absl::make_unique<XdsListenerResourceType>());
-  registry->RegisterType(absl::make_unique<XdsRouteConfigResourceType>());
-  registry->RegisterType(absl::make_unique<XdsClusterResourceType>());
-  registry->RegisterType(absl::make_unique<XdsEndpointResourceType>());
-}
-
-}  // namespace
-
 void XdsClientGlobalInit() {
   g_mu = new Mutex;
   XdsHttpFilterRegistry::Init();
-  static gpr_once once = GPR_ONCE_INIT;
-  gpr_once_init(&once, InitResourceTypeRegistry);
 }
 
 // TODO(roth): Find a better way to clear the fallback config that does

src/core/ext/xds/xds_client.h

@@ -28,10 +28,7 @@
 #include "src/core/ext/xds/xds_api.h"
 #include "src/core/ext/xds/xds_bootstrap.h"
 #include "src/core/ext/xds/xds_client_stats.h"
-#include "src/core/ext/xds/xds_cluster.h"
+#include "src/core/ext/xds/xds_resource_type.h"
-#include "src/core/ext/xds/xds_endpoint.h"
-#include "src/core/ext/xds/xds_listener.h"
-#include "src/core/ext/xds/xds_route_config.h"
 #include "src/core/lib/channel/channelz.h"
 #include "src/core/lib/gprpp/dual_ref_counted.h"
 #include "src/core/lib/gprpp/memory.h"
@@ -39,6 +36,7 @@
 #include "src/core/lib/gprpp/ref_counted.h"
 #include "src/core/lib/gprpp/ref_counted_ptr.h"
 #include "src/core/lib/gprpp/sync.h"
+#include "src/core/lib/iomgr/work_serializer.h"
 
 namespace grpc_core {
 
@@ -48,6 +46,9 @@ extern TraceFlag grpc_xds_client_refcount_trace;
 class XdsClient : public DualRefCounted<XdsClient> {
  public:
   // Resource watcher interface.  Implemented by callers.
+  // Note: Most callers will not use this API directly but rather via a
+  // resource-type-specific wrapper API provided by the relevant
+  // XdsResourceType implementation.
   class ResourceWatcherInterface : public RefCounted<ResourceWatcherInterface> {
    public:
     virtual void OnResourceChanged(
@@ -59,67 +60,6 @@ class XdsClient : public DualRefCounted<XdsClient> {
         ABSL_EXCLUSIVE_LOCKS_REQUIRED(&work_serializer_) = 0;
   };
 
-  // TODO(roth): Consider removing these resource-type-specific APIs in
-  // favor of some mechanism for automatic type-deduction for the generic
-  // API.
-
-  // Listener data watcher interface.  Implemented by callers.
-  class ListenerWatcherInterface : public ResourceWatcherInterface {
-   public:
-    virtual void OnListenerChanged(XdsListenerResource listener) = 0;
-
-   private:
-    void OnResourceChanged(
-        const XdsResourceType::ResourceData* resource) override {
-      OnListenerChanged(
-          static_cast<const XdsListenerResourceType::ListenerData*>(resource)
-              ->resource);
-    }
-  };
-
-  // RouteConfiguration data watcher interface.  Implemented by callers.
-  class RouteConfigWatcherInterface : public ResourceWatcherInterface {
-   public:
-    virtual void OnRouteConfigChanged(XdsRouteConfigResource route_config) = 0;
-
-   private:
-    void OnResourceChanged(
-        const XdsResourceType::ResourceData* resource) override {
-      OnRouteConfigChanged(
-          static_cast<const XdsRouteConfigResourceType::RouteConfigData*>(
-              resource)
-              ->resource);
-    }
-  };
-
-  // Cluster data watcher interface.  Implemented by callers.
-  class ClusterWatcherInterface : public ResourceWatcherInterface {
-   public:
-    virtual void OnClusterChanged(XdsClusterResource cluster_data) = 0;
-
-   private:
-    void OnResourceChanged(
-        const XdsResourceType::ResourceData* resource) override {
-      OnClusterChanged(
-          static_cast<const XdsClusterResourceType::ClusterData*>(resource)
-              ->resource);
-    }
-  };
-
-  // Endpoint data watcher interface.  Implemented by callers.
-  class EndpointWatcherInterface : public ResourceWatcherInterface {
-   public:
-    virtual void OnEndpointChanged(XdsEndpointResource update) = 0;
-
-   private:
-    void OnResourceChanged(
-        const XdsResourceType::ResourceData* resource) override {
-      OnEndpointChanged(
-          static_cast<const XdsEndpointResourceType::EndpointData*>(resource)
-              ->resource);
-    }
-  };
-
   // Factory function to get or create the global XdsClient instance.
   // If *error is not GRPC_ERROR_NONE upon return, then there was
   // an error initializing the client.
@@ -146,12 +86,24 @@ class XdsClient : public DualRefCounted<XdsClient> {
   void Orphan() override;
 
   // Start and cancel watch for a resource.
+  //
   // The XdsClient takes ownership of the watcher, but the caller may
   // keep a raw pointer to the watcher, which may be used only for
   // cancellation.  (Because the caller does not own the watcher, the
   // pointer must not be used for any other purpose.)
   // If the caller is going to start a new watch after cancelling the
   // old one, it should set delay_unsubscription to true.
+  //
+  // The resource type object must be a global singleton, since the first
+  // time the XdsClient sees a particular resource type object, it will
+  // store the pointer to that object as the authoritative implementation for
+  // its type URLs.  The resource type object must outlive the XdsClient object,
+  // and it is illegal to start a subsequent watch for the same type URLs using
+  // a different resource type object.
+  //
+  // Note: Most callers will not use this API directly but rather via a
+  // resource-type-specific wrapper API provided by the relevant
+  // XdsResourceType implementation.
   void WatchResource(const XdsResourceType* type, absl::string_view name,
                      RefCountedPtr<ResourceWatcherInterface> watcher);
   void CancelResourceWatch(const XdsResourceType* type,
@@ -159,58 +111,6 @@ class XdsClient : public DualRefCounted<XdsClient> {
                            ResourceWatcherInterface* watcher,
                            bool delay_unsubscription = false);
 
-  // Start and cancel listener data watch for a listener.
-  // The XdsClient takes ownership of the watcher, but the caller may
-  // keep a raw pointer to the watcher, which may be used only for
-  // cancellation.  (Because the caller does not own the watcher, the
-  // pointer must not be used for any other purpose.)
-  // If the caller is going to start a new watch after cancelling the
-  // old one, it should set delay_unsubscription to true.
-  void WatchListenerData(absl::string_view listener_name,
-                         RefCountedPtr<ListenerWatcherInterface> watcher);
-  void CancelListenerDataWatch(absl::string_view listener_name,
-                               ListenerWatcherInterface* watcher,
-                               bool delay_unsubscription = false);
-
-  // Start and cancel route config data watch for a listener.
-  // The XdsClient takes ownership of the watcher, but the caller may
-  // keep a raw pointer to the watcher, which may be used only for
-  // cancellation.  (Because the caller does not own the watcher, the
-  // pointer must not be used for any other purpose.)
-  // If the caller is going to start a new watch after cancelling the
-  // old one, it should set delay_unsubscription to true.
-  void WatchRouteConfigData(absl::string_view route_config_name,
-                            RefCountedPtr<RouteConfigWatcherInterface> watcher);
-  void CancelRouteConfigDataWatch(absl::string_view route_config_name,
-                                  RouteConfigWatcherInterface* watcher,
-                                  bool delay_unsubscription = false);
-
-  // Start and cancel cluster data watch for a cluster.
-  // The XdsClient takes ownership of the watcher, but the caller may
-  // keep a raw pointer to the watcher, which may be used only for
-  // cancellation.  (Because the caller does not own the watcher, the
-  // pointer must not be used for any other purpose.)
-  // If the caller is going to start a new watch after cancelling the
-  // old one, it should set delay_unsubscription to true.
-  void WatchClusterData(absl::string_view cluster_name,
-                        RefCountedPtr<ClusterWatcherInterface> watcher);
-  void CancelClusterDataWatch(absl::string_view cluster_name,
-                              ClusterWatcherInterface* watcher,
-                              bool delay_unsubscription = false);
-
-  // Start and cancel endpoint data watch for a cluster.
-  // The XdsClient takes ownership of the watcher, but the caller may
-  // keep a raw pointer to the watcher, which may be used only for
-  // cancellation.  (Because the caller does not own the watcher, the
-  // pointer must not be used for any other purpose.)
-  // If the caller is going to start a new watch after cancelling the
-  // old one, it should set delay_unsubscription to true.
-  void WatchEndpointData(absl::string_view eds_service_name,
-                         RefCountedPtr<EndpointWatcherInterface> watcher);
-  void CancelEndpointDataWatch(absl::string_view eds_service_name,
-                               EndpointWatcherInterface* watcher,
-                               bool delay_unsubscription = false);
-
   // Adds and removes drop stats for cluster_name and eds_service_name.
   RefCountedPtr<XdsClusterDropStats> AddClusterDropStats(
       absl::string_view lrs_server, absl::string_view cluster_name,
@@ -350,6 +250,13 @@ class XdsClient : public DualRefCounted<XdsClient> {
   void NotifyOnErrorLocked(grpc_error_handle error)
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
+  void MaybeRegisterResourceTypeLocked(const XdsResourceType* resource_type)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
+
+  // Gets the type for resource_type, or null if the type is unknown.
+  const XdsResourceType* GetResourceTypeLocked(absl::string_view resource_type)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
+
   static absl::StatusOr<XdsResourceName> ParseXdsResourceName(
       absl::string_view name, const XdsResourceType* type);
   static std::string ConstructFullXdsResourceName(
@@ -373,6 +280,13 @@ class XdsClient : public DualRefCounted<XdsClient> {
 
   Mutex mu_;
 
+  // Stores resource type objects seen by type URL.
+  std::map<absl::string_view /*resource_type*/, const XdsResourceType*>
+      resource_types_ ABSL_GUARDED_BY(mu_);
+  std::map<absl::string_view /*v2_resource_type*/, const XdsResourceType*>
+      v2_resource_types_ ABSL_GUARDED_BY(mu_);
+  upb::SymbolTable symtab_ ABSL_GUARDED_BY(mu_);
+
   //  Map of existing xDS server channels.
   std::map<XdsBootstrap::XdsServer, ChannelState*> xds_server_channel_map_
       ABSL_GUARDED_BY(mu_);

src/core/ext/xds/xds_cluster.h

@@ -27,6 +27,7 @@
 #include "envoy/extensions/clusters/aggregate/v3/cluster.upbdefs.h"
 #include "envoy/extensions/transport_sockets/tls/v3/tls.upbdefs.h"
 
+#include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_common_types.h"
 #include "src/core/ext/xds/xds_resource_type.h"
 
@@ -87,6 +88,25 @@ class XdsClusterResourceType : public XdsResourceType {
     XdsClusterResource resource;
   };
 
+  class WatcherInterface : public XdsClient::ResourceWatcherInterface {
+   public:
+    virtual void OnClusterChanged(XdsClusterResource cluster_data) = 0;
+
+   private:
+    void OnResourceChanged(
+        const XdsResourceType::ResourceData* resource) override {
+      OnClusterChanged(
+          static_cast<const XdsClusterResourceType::ClusterData*>(resource)
+              ->resource);
+    }
+  };
+
+  static const XdsClusterResourceType* Get() {
+    static const XdsClusterResourceType* g_instance =
+        new XdsClusterResourceType();
+    return g_instance;
+  }
+
   absl::string_view type_url() const override {
     return "envoy.config.cluster.v3.Cluster";
   }
@@ -94,6 +114,19 @@ class XdsClusterResourceType : public XdsResourceType {
     return "envoy.api.v2.Cluster";
   }
 
+  static void StartWatch(XdsClient* xds_client, absl::string_view resource_name,
+                         RefCountedPtr<WatcherInterface> watcher) {
+    xds_client->WatchResource(Get(), resource_name, std::move(watcher));
+  }
+
+  static void CancelWatch(XdsClient* xds_client,
+                          absl::string_view resource_name,
+                          WatcherInterface* watcher,
+                          bool delay_unsubscription = false) {
+    xds_client->CancelResourceWatch(Get(), resource_name, watcher,
+                                    delay_unsubscription);
+  }
+
   absl::StatusOr<DecodeResult> Decode(const XdsEncodingContext& context,
                                       absl::string_view serialized_resource,
                                       bool is_v2) const override;

src/core/ext/xds/xds_endpoint.h

@@ -27,6 +27,7 @@
 #include "envoy/config/endpoint/v3/endpoint.upbdefs.h"
 
 #include "src/core/ext/filters/client_channel/server_address.h"
+#include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_client_stats.h"
 #include "src/core/ext/xds/xds_resource_type.h"
 #include "src/core/lib/gprpp/ref_counted_ptr.h"
@@ -116,6 +117,25 @@ class XdsEndpointResourceType : public XdsResourceType {
     XdsEndpointResource resource;
   };
 
+  class WatcherInterface : public XdsClient::ResourceWatcherInterface {
+   public:
+    virtual void OnEndpointChanged(XdsEndpointResource update) = 0;
+
+   private:
+    void OnResourceChanged(
+        const XdsResourceType::ResourceData* resource) override {
+      OnEndpointChanged(
+          static_cast<const XdsEndpointResourceType::EndpointData*>(resource)
+              ->resource);
+    }
+  };
+
+  static const XdsEndpointResourceType* Get() {
+    static const XdsEndpointResourceType* g_instance =
+        new XdsEndpointResourceType();
+    return g_instance;
+  }
+
   absl::string_view type_url() const override {
     return "envoy.config.endpoint.v3.ClusterLoadAssignment";
   }
@@ -123,6 +143,19 @@ class XdsEndpointResourceType : public XdsResourceType {
     return "envoy.api.v2.ClusterLoadAssignment";
   }
 
+  static void StartWatch(XdsClient* xds_client, absl::string_view resource_name,
+                         RefCountedPtr<WatcherInterface> watcher) {
+    xds_client->WatchResource(Get(), resource_name, std::move(watcher));
+  }
+
+  static void CancelWatch(XdsClient* xds_client,
+                          absl::string_view resource_name,
+                          WatcherInterface* watcher,
+                          bool delay_unsubscription = false) {
+    xds_client->CancelResourceWatch(Get(), resource_name, watcher,
+                                    delay_unsubscription);
+  }
+
   absl::StatusOr<DecodeResult> Decode(const XdsEncodingContext& context,
                                       absl::string_view serialized_resource,
                                       bool is_v2) const override;

src/core/ext/xds/xds_listener.h

@@ -30,6 +30,7 @@
 #include "envoy/config/listener/v3/listener.upbdefs.h"
 #include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.upbdefs.h"
 
+#include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_common_types.h"
 #include "src/core/ext/xds/xds_http_filters.h"
 #include "src/core/ext/xds/xds_route_config.h"
@@ -195,6 +196,25 @@ class XdsListenerResourceType : public XdsResourceType {
     XdsListenerResource resource;
   };
 
+  class WatcherInterface : public XdsClient::ResourceWatcherInterface {
+   public:
+    virtual void OnListenerChanged(XdsListenerResource listener) = 0;
+
+   private:
+    void OnResourceChanged(
+        const XdsResourceType::ResourceData* resource) override {
+      OnListenerChanged(
+          static_cast<const XdsListenerResourceType::ListenerData*>(resource)
+              ->resource);
+    }
+  };
+
+  static const XdsListenerResourceType* Get() {
+    static const XdsListenerResourceType* g_instance =
+        new XdsListenerResourceType();
+    return g_instance;
+  }
+
   absl::string_view type_url() const override {
     return "envoy.config.listener.v3.Listener";
   }
@@ -202,6 +222,19 @@ class XdsListenerResourceType : public XdsResourceType {
     return "envoy.api.v2.Listener";
   }
 
+  static void StartWatch(XdsClient* xds_client, absl::string_view resource_name,
+                         RefCountedPtr<WatcherInterface> watcher) {
+    xds_client->WatchResource(Get(), resource_name, std::move(watcher));
+  }
+
+  static void CancelWatch(XdsClient* xds_client,
+                          absl::string_view resource_name,
+                          WatcherInterface* watcher,
+                          bool delay_unsubscription = false) {
+    xds_client->CancelResourceWatch(Get(), resource_name, watcher,
+                                    delay_unsubscription);
+  }
+
   absl::StatusOr<DecodeResult> Decode(const XdsEncodingContext& context,
                                       absl::string_view serialized_resource,
                                       bool is_v2) const override;

src/core/ext/xds/xds_resource_type.cc

@@ -18,13 +18,6 @@
 
 #include "src/core/ext/xds/xds_resource_type.h"
 
-#include <vector>
-
-#include "absl/strings/str_join.h"
-#include "absl/strings/str_split.h"
-
-#include "src/core/lib/uri/uri_parser.h"
-
 namespace grpc_core {
 
 bool XdsResourceType::IsType(absl::string_view resource_type,
@@ -37,35 +30,4 @@ bool XdsResourceType::IsType(absl::string_view resource_type,
   return false;
 }
 
-XdsResourceTypeRegistry* XdsResourceTypeRegistry::GetOrCreate() {
-  static XdsResourceTypeRegistry* registry = new XdsResourceTypeRegistry();
-  return registry;
-}
-
-const XdsResourceType* XdsResourceTypeRegistry::GetType(
-    absl::string_view resource_type) {
-  auto it = resource_types_.find(resource_type);
-  if (it != resource_types_.end()) return it->second.get();
-  auto it2 = v2_resource_types_.find(resource_type);
-  if (it2 != v2_resource_types_.end()) return it2->second;
-  return nullptr;
-}
-
-void XdsResourceTypeRegistry::RegisterType(
-    std::unique_ptr<XdsResourceType> resource_type) {
-  GPR_ASSERT(resource_types_.find(resource_type->type_url()) ==
-             resource_types_.end());
-  GPR_ASSERT(v2_resource_types_.find(resource_type->v2_type_url()) ==
-             v2_resource_types_.end());
-  v2_resource_types_.emplace(resource_type->v2_type_url(), resource_type.get());
-  resource_types_.emplace(resource_type->type_url(), std::move(resource_type));
-}
-
-void XdsResourceTypeRegistry::ForEach(
-    std::function<void(const XdsResourceType*)> func) {
-  for (const auto& p : resource_types_) {
-    func(p.second.get());
-  }
-}
-
 }  // namespace grpc_core

src/core/ext/xds/xds_resource_type.h

@@ -16,7 +16,6 @@
 
 #include <grpc/support/port_platform.h>
 
-#include <map>
 #include <memory>
 #include <string>
 
@@ -94,30 +93,6 @@ class XdsResourceType {
   bool IsType(absl::string_view resource_type, bool* is_v2) const;
 };
 
-// Global registry of xDS resource types.
-class XdsResourceTypeRegistry {
- public:
-  // Gets the global registry, creating it if needed.
-  static XdsResourceTypeRegistry* GetOrCreate();
-
-  // Gets the type for resource_type, or null if the type is unknown.
-  const XdsResourceType* GetType(absl::string_view resource_type);
-
-  // Registers a resource type.
-  // All types must be registered before they can be used in the XdsClient.
-  void RegisterType(std::unique_ptr<XdsResourceType> resource_type);
-
-  // Calls func for each resource type.
-  void ForEach(std::function<void(const XdsResourceType*)> func);
-
- private:
-  std::map<absl::string_view /*resource_type*/,
-           std::unique_ptr<XdsResourceType>>
-      resource_types_;
-  std::map<absl::string_view /*v2_resource_type*/, XdsResourceType*>
-      v2_resource_types_;
-};
-
 }  // namespace grpc_core
 
 #endif  // GRPC_CORE_EXT_XDS_XDS_RESOURCE_TYPE_H

src/core/ext/xds/xds_route_config.h

@@ -29,6 +29,7 @@
 #include "envoy/config/route/v3/route.upbdefs.h"
 #include "re2/re2.h"
 
+#include "src/core/ext/xds/xds_client.h"
 #include "src/core/ext/xds/xds_common_types.h"
 #include "src/core/ext/xds/xds_http_filters.h"
 #include "src/core/ext/xds/xds_resource_type.h"
@@ -195,6 +196,26 @@ class XdsRouteConfigResourceType : public XdsResourceType {
     XdsRouteConfigResource resource;
   };
 
+  class WatcherInterface : public XdsClient::ResourceWatcherInterface {
+   public:
+    virtual void OnRouteConfigChanged(XdsRouteConfigResource route_config) = 0;
+
+   private:
+    void OnResourceChanged(
+        const XdsResourceType::ResourceData* resource) override {
+      OnRouteConfigChanged(
+          static_cast<const XdsRouteConfigResourceType::RouteConfigData*>(
+              resource)
+              ->resource);
+    }
+  };
+
+  static const XdsRouteConfigResourceType* Get() {
+    static const XdsRouteConfigResourceType* g_instance =
+        new XdsRouteConfigResourceType();
+    return g_instance;
+  }
+
   absl::string_view type_url() const override {
     return "envoy.config.route.v3.RouteConfiguration";
   }
@@ -202,6 +223,19 @@ class XdsRouteConfigResourceType : public XdsResourceType {
     return "envoy.api.v2.RouteConfiguration";
   }
 
+  static void StartWatch(XdsClient* xds_client, absl::string_view resource_name,
+                         RefCountedPtr<WatcherInterface> watcher) {
+    xds_client->WatchResource(Get(), resource_name, std::move(watcher));
+  }
+
+  static void CancelWatch(XdsClient* xds_client,
+                          absl::string_view resource_name,
+                          WatcherInterface* watcher,
+                          bool delay_unsubscription = false) {
+    xds_client->CancelResourceWatch(Get(), resource_name, watcher,
+                                    delay_unsubscription);
+  }
+
   absl::StatusOr<DecodeResult> Decode(const XdsEncodingContext& context,
                                       absl::string_view serialized_resource,
                                       bool /*is_v2*/) const override;

src/core/ext/xds/xds_server_config_fetcher.cc

@@ -26,6 +26,8 @@
 #include "src/core/ext/xds/xds_certificate_provider.h"
 #include "src/core/ext/xds/xds_channel_stack_modifier.h"
 #include "src/core/ext/xds/xds_client.h"
+#include "src/core/ext/xds/xds_listener.h"
+#include "src/core/ext/xds/xds_route_config.h"
 #include "src/core/ext/xds/xds_routing.h"
 #include "src/core/lib/address_utils/sockaddr_utils.h"
 #include "src/core/lib/channel/channel_args.h"
@@ -86,7 +88,7 @@ class XdsServerConfigFetcher : public grpc_server_config_fetcher {
 // update received was a fatal error (resource does not exist), the server
 // listener is made to stop listening.
 class XdsServerConfigFetcher::ListenerWatcher
-    : public XdsClient::ListenerWatcherInterface {
+    : public XdsListenerResourceType::WatcherInterface {
  public:
   ListenerWatcher(RefCountedPtr<XdsClient> xds_client,
                   std::unique_ptr<grpc_server_config_fetcher::WatcherInterface>
@@ -217,7 +219,7 @@ class XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager
 // with the latest updates and new connections do not need to wait for the RDS
 // resources to be fetched.
 class XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
-    RouteConfigWatcher : public XdsClient::RouteConfigWatcherInterface {
+    RouteConfigWatcher : public XdsRouteConfigResourceType::WatcherInterface {
  public:
   RouteConfigWatcher(
       std::string resource_name,
@@ -381,7 +383,7 @@ class XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
 // DynamicXdsServerConfigSelectorProvider
 class XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
     DynamicXdsServerConfigSelectorProvider::RouteConfigWatcher
-    : public XdsClient::RouteConfigWatcherInterface {
+    : public XdsRouteConfigResourceType::WatcherInterface {
  public:
   explicit RouteConfigWatcher(
       RefCountedPtr<DynamicXdsServerConfigSelectorProvider> parent)
@@ -418,7 +420,8 @@ void XdsServerConfigFetcher::StartWatch(
       xds_client_, std::move(watcher), serving_status_notifier_,
       listening_address);
   auto* listener_watcher_ptr = listener_watcher.get();
-  xds_client_->WatchListenerData(
+  XdsListenerResourceType::StartWatch(
+      xds_client_.get(),
       absl::StrReplaceAll(
           xds_client_->bootstrap().server_listener_resource_name_template(),
           {{"%s", listening_address}}),
@@ -433,7 +436,8 @@ void XdsServerConfigFetcher::CancelWatch(
   auto it = listener_watchers_.find(watcher);
   if (it != listener_watchers_.end()) {
     // Cancel the watch on the listener before erasing
-    xds_client_->CancelListenerDataWatch(
+    XdsListenerResourceType::CancelWatch(
+        xds_client_.get(),
         absl::StrReplaceAll(
             xds_client_->bootstrap().server_listener_resource_name_template(),
             {{"%s", it->second->listening_address()}}),
@@ -618,8 +622,8 @@ void XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
           MakeRefCounted<RouteConfigWatcher>(resource_name, WeakRef());
       rds_map_.emplace(resource_name, RdsUpdateState{route_config_watcher.get(),
                                                      absl::nullopt});
-      xds_client_->WatchRouteConfigData(resource_name,
+      XdsRouteConfigResourceType::StartWatch(xds_client_.get(), resource_name,
-                                        std::move(route_config_watcher));
+                                             std::move(route_config_watcher));
     }
     if (rds_resources_yet_to_fetch_ != 0) {
       listener_watcher_ = std::move(listener_watcher);
@@ -638,7 +642,8 @@ void XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
   MutexLock lock(&mu_);
   // Cancel the RDS watches to clear up the weak refs
   for (const auto& entry : rds_map_) {
-    xds_client_->CancelRouteConfigDataWatch(entry.first, entry.second.watcher,
+    XdsRouteConfigResourceType::CancelWatch(xds_client_.get(), entry.first,
+                                            entry.second.watcher,
                                             false /* delay_unsubscription */);
   }
   // Also give up the ref on ListenerWatcher since it won't be needed anymore
@@ -1159,8 +1164,8 @@ XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
   GPR_ASSERT(!resource_name_.empty());
   auto route_config_watcher = MakeRefCounted<RouteConfigWatcher>(Ref());
   route_config_watcher_ = route_config_watcher.get();
-  xds_client_->WatchRouteConfigData(resource_name_,
+  XdsRouteConfigResourceType::StartWatch(xds_client_.get(), resource_name_,
-                                    std::move(route_config_watcher));
+                                         std::move(route_config_watcher));
 }
 
 absl::StatusOr<RefCountedPtr<ServerConfigSelector>>
@@ -1184,7 +1189,8 @@ XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
 
 void XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
     DynamicXdsServerConfigSelectorProvider::CancelWatch() {
-  xds_client_->CancelRouteConfigDataWatch(resource_name_, route_config_watcher_,
+  XdsRouteConfigResourceType::CancelWatch(xds_client_.get(), resource_name_,
+                                          route_config_watcher_,
                                           false /* delay_unsubscription */);
   MutexLock lock(&mu_);
   watcher_.reset();

test/cpp/end2end/xds/xds_end2end_test.cc

@@ -64,6 +64,7 @@
 #include "src/core/ext/xds/xds_api.h"
 #include "src/core/ext/xds/xds_channel_args.h"
 #include "src/core/ext/xds/xds_client.h"
+#include "src/core/ext/xds/xds_listener.h"
 #include "src/core/lib/address_utils/parse_address.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/gpr/env.h"