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

This reverts commit 747a5d8c24.

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

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

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

@@ -30,7 +30,6 @@
 #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,7 +39,6 @@
 #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"
@@ -166,7 +165,7 @@ class XdsClusterResolverLb : public LoadBalancingPolicy {
     bool disable_reresolution() override { return true; }
 
    private:
-    class EndpointWatcher : public XdsEndpointResourceType::WatcherInterface {
+    class EndpointWatcher : public XdsClient::EndpointWatcherInterface {
      public:
       explicit EndpointWatcher(
           RefCountedPtr<EdsDiscoveryMechanism> discovery_mechanism)
@@ -415,8 +414,8 @@ void XdsClusterResolverLb::EdsDiscoveryMechanism::Start() {
   auto watcher = MakeRefCounted<EndpointWatcher>(
       Ref(DEBUG_LOCATION, "EdsDiscoveryMechanism"));
   watcher_ = watcher.get();
-  XdsEndpointResourceType::StartWatch(parent()->xds_client_.get(),
+  parent()->xds_client_->WatchEndpointData(GetEdsResourceName(),
-                                      GetEdsResourceName(), std::move(watcher));
+                                           std::move(watcher));
 }
 
 void XdsClusterResolverLb::EdsDiscoveryMechanism::Orphan() {
@@ -426,8 +425,8 @@ void XdsClusterResolverLb::EdsDiscoveryMechanism::Orphan() {
             ":%p cancelling xds watch for %s",
             parent(), index(), this, std::string(GetEdsResourceName()).c_str());
   }
-  XdsEndpointResourceType::CancelWatch(parent()->xds_client_.get(),
+  parent()->xds_client_->CancelEndpointDataWatch(GetEdsResourceName(),
-                                       GetEdsResourceName(), watcher_);
+                                                 watcher_);
   Unref();
 }
 

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

@@ -31,8 +31,6 @@
 #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"
@@ -82,7 +80,7 @@ class XdsResolver : public Resolver {
   }
 
  private:
-  class ListenerWatcher : public XdsListenerResourceType::WatcherInterface {
+  class ListenerWatcher : public XdsClient::ListenerWatcherInterface {
    public:
     explicit ListenerWatcher(RefCountedPtr<XdsResolver> resolver)
         : resolver_(std::move(resolver)) {}
@@ -120,8 +118,7 @@ class XdsResolver : public Resolver {
     RefCountedPtr<XdsResolver> resolver_;
   };
 
-  class RouteConfigWatcher
+  class RouteConfigWatcher : public XdsClient::RouteConfigWatcherInterface {
-      : public XdsRouteConfigResourceType::WatcherInterface {
    public:
     explicit RouteConfigWatcher(RefCountedPtr<XdsResolver> resolver)
         : resolver_(std::move(resolver)) {}
@@ -294,14 +291,14 @@ class XdsResolver : public Resolver {
 
   RefCountedPtr<XdsClient> xds_client_;
 
-  ListenerWatcher* listener_watcher_ = nullptr;
+  XdsClient::ListenerWatcherInterface* 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_;
-  RouteConfigWatcher* route_config_watcher_ = nullptr;
+  XdsClient::RouteConfigWatcherInterface* route_config_watcher_ = nullptr;
   XdsRouteConfigResource::VirtualHost current_virtual_host_;
 
   ClusterState::ClusterStateMap cluster_state_map_;
@@ -697,8 +694,7 @@ void XdsResolver::StartLocked() {
                                    interested_parties_);
   auto watcher = MakeRefCounted<ListenerWatcher>(Ref());
   listener_watcher_ = watcher.get();
-  XdsListenerResourceType::StartWatch(xds_client_.get(), server_name_,
+  xds_client_->WatchListenerData(server_name_, std::move(watcher));
-                                      std::move(watcher));
 }
 
 void XdsResolver::ShutdownLocked() {
@@ -707,14 +703,12 @@ void XdsResolver::ShutdownLocked() {
   }
   if (xds_client_ != nullptr) {
     if (listener_watcher_ != nullptr) {
-      XdsListenerResourceType::CancelWatch(xds_client_.get(), server_name_,
+      xds_client_->CancelListenerDataWatch(server_name_, listener_watcher_,
-                                           listener_watcher_,
                                            /*delay_unsubscription=*/false);
     }
     if (route_config_watcher_ != nullptr) {
-      XdsRouteConfigResourceType::CancelWatch(
+      xds_client_->CancelRouteConfigDataWatch(
-          xds_client_.get(), route_config_name_, route_config_watcher_,
+          server_name_, route_config_watcher_, /*delay_unsubscription=*/false);
-          /*delay_unsubscription=*/false);
     }
     grpc_pollset_set_del_pollset_set(xds_client_->interested_parties(),
                                      interested_parties_);
@@ -732,8 +726,8 @@ void XdsResolver::OnListenerUpdate(XdsListenerResource listener) {
   if (listener.http_connection_manager.route_config_name !=
       route_config_name_) {
     if (route_config_watcher_ != nullptr) {
-      XdsRouteConfigResourceType::CancelWatch(
+      xds_client_->CancelRouteConfigDataWatch(
-          xds_client_.get(), route_config_name_, route_config_watcher_,
+          route_config_name_, route_config_watcher_,
           /*delay_unsubscription=*/
           !listener.http_connection_manager.route_config_name.empty());
       route_config_watcher_ = nullptr;
@@ -744,8 +738,7 @@ void XdsResolver::OnListenerUpdate(XdsListenerResource listener) {
       current_virtual_host_.routes.clear();
       auto watcher = MakeRefCounted<RouteConfigWatcher>(Ref());
       route_config_watcher_ = watcher.get();
-      XdsRouteConfigResourceType::StartWatch(
+      xds_client_->WatchRouteConfigData(route_config_name_, std::move(watcher));
-          xds_client_.get(), route_config_name_, std::move(watcher));
     }
   }
   current_listener_ = std::move(listener);

src/core/ext/xds/xds_client.cc

@@ -37,10 +37,7 @@
 #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"
@@ -79,6 +76,11 @@ 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 {
@@ -1922,6 +1924,66 @@ void XdsClient::CancelResourceWatch(const XdsResourceType* type,
   }
 }
 
+void XdsClient::WatchListenerData(
+    absl::string_view listener_name,
+    RefCountedPtr<ListenerWatcherInterface> watcher) {
+  WatchResource(XdsResourceTypeRegistry::GetOrCreate()->GetType(kLdsTypeUrl),
+                listener_name, std::move(watcher));
+}
+
+void XdsClient::CancelListenerDataWatch(absl::string_view listener_name,
+                                        ListenerWatcherInterface* watcher,
+                                        bool delay_unsubscription) {
+  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,
+                                        EndpointWatcherInterface* watcher,
+                                        bool delay_unsubscription) {
+  CancelResourceWatch(
+      XdsResourceTypeRegistry::GetOrCreate()->GetType(kEdsTypeUrl),
+      eds_service_name, watcher, delay_unsubscription);
+}
+
 absl::StatusOr<XdsClient::XdsResourceName> XdsClient::ParseXdsResourceName(
     absl::string_view name, const XdsResourceType* type) {
   // Old-style names use the empty string for authority.
@@ -1988,8 +2050,9 @@ RefCountedPtr<XdsClusterDropStats> XdsClient::AddClusterDropStats(
         it->first.second /*eds_service_name*/);
     load_report_state.drop_stats = cluster_drop_stats.get();
   }
-  auto resource_name =
+  auto resource_name = ParseXdsResourceName(
-      ParseXdsResourceName(cluster_name, XdsClusterResourceType::Get());
+      cluster_name,
+      XdsResourceTypeRegistry::GetOrCreate()->GetType(kCdsTypeUrl));
   GPR_ASSERT(resource_name.ok());
   auto a = authority_state_map_.find(resource_name->authority);
   if (a != authority_state_map_.end()) {
@@ -2051,8 +2114,9 @@ RefCountedPtr<XdsClusterLocalityStats> XdsClient::AddClusterLocalityStats(
         std::move(locality));
     locality_state.locality_stats = cluster_locality_stats.get();
   }
-  auto resource_name =
+  auto resource_name = ParseXdsResourceName(
-      ParseXdsResourceName(cluster_name, XdsClusterResourceType::Get());
+      cluster_name,
+      XdsResourceTypeRegistry::GetOrCreate()->GetType(kCdsTypeUrl));
   GPR_ASSERT(resource_name.ok());
   auto a = authority_state_map_.find(resource_name->authority);
   if (a != authority_state_map_.end()) {
@@ -2221,9 +2285,23 @@ 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,7 +28,10 @@
 #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_resource_type.h"
+#include "src/core/ext/xds/xds_cluster.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"
@@ -36,7 +39,6 @@
 #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 {
 
@@ -57,6 +59,67 @@ 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.
@@ -96,6 +159,58 @@ 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,

src/core/ext/xds/xds_cluster.h

@@ -27,7 +27,6 @@
 #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"
 
@@ -88,25 +87,6 @@ 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";
   }
@@ -114,19 +94,6 @@ 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;
@@ -153,11 +120,6 @@ class XdsClusterResourceType : public XdsResourceType {
     envoy_extensions_transport_sockets_tls_v3_UpstreamTlsContext_getmsgdef(
         symtab);
   }
-
- private:
-  XdsClusterResourceType() {
-    XdsResourceTypeRegistry::GetOrCreate()->RegisterType(this);
-  }
 };
 
 }  // namespace grpc_core

src/core/ext/xds/xds_endpoint.h

@@ -27,7 +27,6 @@
 #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"
@@ -117,25 +116,6 @@ 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";
   }
@@ -143,19 +123,6 @@ 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;
@@ -177,11 +144,6 @@ class XdsEndpointResourceType : public XdsResourceType {
   void InitUpbSymtab(upb_symtab* symtab) const override {
     envoy_config_endpoint_v3_ClusterLoadAssignment_getmsgdef(symtab);
   }
-
- private:
-  XdsEndpointResourceType() {
-    XdsResourceTypeRegistry::GetOrCreate()->RegisterType(this);
-  }
 };
 
 }  // namespace grpc_core

src/core/ext/xds/xds_listener.h

@@ -30,7 +30,6 @@
 #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"
@@ -196,25 +195,6 @@ 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";
   }
@@ -222,19 +202,6 @@ 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;
@@ -261,11 +228,6 @@ class XdsListenerResourceType : public XdsResourceType {
         symtab);
     XdsHttpFilterRegistry::PopulateSymtab(symtab);
   }
-
- private:
-  XdsListenerResourceType() {
-    XdsResourceTypeRegistry::GetOrCreate()->RegisterType(this);
-  }
 };
 
 }  // namespace grpc_core

src/core/ext/xds/xds_resource_type.cc

@@ -45,26 +45,26 @@ XdsResourceTypeRegistry* XdsResourceTypeRegistry::GetOrCreate() {
 const XdsResourceType* XdsResourceTypeRegistry::GetType(
     absl::string_view resource_type) {
   auto it = resource_types_.find(resource_type);
-  if (it != resource_types_.end()) return it->second;
+  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(
-    const XdsResourceType* resource_type) {
+    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);
+  v2_resource_types_.emplace(resource_type->v2_type_url(), resource_type.get());
-  resource_types_.emplace(resource_type->type_url(), resource_type);
+  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);
+    func(p.second.get());
   }
 }
 

src/core/ext/xds/xds_resource_type.h

@@ -105,15 +105,16 @@ class XdsResourceTypeRegistry {
 
   // Registers a resource type.
   // All types must be registered before they can be used in the XdsClient.
-  void RegisterType(const XdsResourceType* resource_type);
+  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*/, const XdsResourceType*>
+  std::map<absl::string_view /*resource_type*/,
+           std::unique_ptr<XdsResourceType>>
       resource_types_;
-  std::map<absl::string_view /*v2_resource_type*/, const XdsResourceType*>
+  std::map<absl::string_view /*v2_resource_type*/, XdsResourceType*>
       v2_resource_types_;
 };
 

src/core/ext/xds/xds_route_config.h

@@ -29,7 +29,6 @@
 #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"
@@ -196,26 +195,6 @@ 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";
   }
@@ -223,19 +202,6 @@ 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;
@@ -257,11 +223,6 @@ class XdsRouteConfigResourceType : public XdsResourceType {
   void InitUpbSymtab(upb_symtab* symtab) const override {
     envoy_config_route_v3_RouteConfiguration_getmsgdef(symtab);
   }
-
- private:
-  XdsRouteConfigResourceType() {
-    XdsResourceTypeRegistry::GetOrCreate()->RegisterType(this);
-  }
 };
 
 }  // namespace grpc_core

src/core/ext/xds/xds_server_config_fetcher.cc

@@ -26,8 +26,6 @@
 #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"
@@ -88,7 +86,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 XdsListenerResourceType::WatcherInterface {
+    : public XdsClient::ListenerWatcherInterface {
  public:
   ListenerWatcher(RefCountedPtr<XdsClient> xds_client,
                   std::unique_ptr<grpc_server_config_fetcher::WatcherInterface>
@@ -219,7 +217,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 XdsRouteConfigResourceType::WatcherInterface {
+    RouteConfigWatcher : public XdsClient::RouteConfigWatcherInterface {
  public:
   RouteConfigWatcher(
       std::string resource_name,
@@ -383,7 +381,7 @@ class XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
 // DynamicXdsServerConfigSelectorProvider
 class XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
     DynamicXdsServerConfigSelectorProvider::RouteConfigWatcher
-    : public XdsRouteConfigResourceType::WatcherInterface {
+    : public XdsClient::RouteConfigWatcherInterface {
  public:
   explicit RouteConfigWatcher(
       RefCountedPtr<DynamicXdsServerConfigSelectorProvider> parent)
@@ -420,8 +418,7 @@ void XdsServerConfigFetcher::StartWatch(
       xds_client_, std::move(watcher), serving_status_notifier_,
       listening_address);
   auto* listener_watcher_ptr = listener_watcher.get();
-  XdsListenerResourceType::StartWatch(
+  xds_client_->WatchListenerData(
-      xds_client_.get(),
       absl::StrReplaceAll(
           xds_client_->bootstrap().server_listener_resource_name_template(),
           {{"%s", listening_address}}),
@@ -436,8 +433,7 @@ void XdsServerConfigFetcher::CancelWatch(
   auto it = listener_watchers_.find(watcher);
   if (it != listener_watchers_.end()) {
     // Cancel the watch on the listener before erasing
-    XdsListenerResourceType::CancelWatch(
+    xds_client_->CancelListenerDataWatch(
-        xds_client_.get(),
         absl::StrReplaceAll(
             xds_client_->bootstrap().server_listener_resource_name_template(),
             {{"%s", it->second->listening_address()}}),
@@ -622,8 +618,8 @@ void XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
           MakeRefCounted<RouteConfigWatcher>(resource_name, WeakRef());
       rds_map_.emplace(resource_name, RdsUpdateState{route_config_watcher.get(),
                                                      absl::nullopt});
-      XdsRouteConfigResourceType::StartWatch(xds_client_.get(), resource_name,
+      xds_client_->WatchRouteConfigData(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);
@@ -642,8 +638,7 @@ void XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
   MutexLock lock(&mu_);
   // Cancel the RDS watches to clear up the weak refs
   for (const auto& entry : rds_map_) {
-    XdsRouteConfigResourceType::CancelWatch(xds_client_.get(), entry.first,
+    xds_client_->CancelRouteConfigDataWatch(entry.first, entry.second.watcher,
-                                            entry.second.watcher,
                                             false /* delay_unsubscription */);
   }
   // Also give up the ref on ListenerWatcher since it won't be needed anymore
@@ -1164,8 +1159,8 @@ XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
   GPR_ASSERT(!resource_name_.empty());
   auto route_config_watcher = MakeRefCounted<RouteConfigWatcher>(Ref());
   route_config_watcher_ = route_config_watcher.get();
-  XdsRouteConfigResourceType::StartWatch(xds_client_.get(), resource_name_,
+  xds_client_->WatchRouteConfigData(resource_name_,
-                                         std::move(route_config_watcher));
+                                    std::move(route_config_watcher));
 }
 
 absl::StatusOr<RefCountedPtr<ServerConfigSelector>>
@@ -1189,8 +1184,7 @@ XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
 
 void XdsServerConfigFetcher::ListenerWatcher::FilterChainMatchManager::
     DynamicXdsServerConfigSelectorProvider::CancelWatch() {
-  XdsRouteConfigResourceType::CancelWatch(xds_client_.get(), resource_name_,
+  xds_client_->CancelRouteConfigDataWatch(resource_name_, route_config_watcher_,
-                                          route_config_watcher_,
                                           false /* delay_unsubscription */);
   MutexLock lock(&mu_);
   watcher_.reset();

test/cpp/end2end/xds/xds_end2end_test.cc

@@ -64,7 +64,6 @@
 #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"