Xds mux unification: introduced unifed mux implementation

envoy/config/grpc_mux.h

@@ -31,6 +31,11 @@ struct ControlPlaneStats {
                           GENERATE_TEXT_READOUT_STRUCT)
 };
 
+// TODO (dmitri-d) This forward declaration is needed in order for the unified mux to fulfill legacy
+// mux interface. It should be removed together with the rest of legacy mux code when the switch to
+// unified mux has been finalized.
+struct Watch;
+
 /**
  * Handle on a muxed gRPC subscription. The subscription is canceled on destruction.
  */
@@ -105,6 +110,24 @@ class GrpcMux {
 
   virtual void requestOnDemandUpdate(const std::string& type_url,
                                      const absl::flat_hash_set<std::string>& for_update) PURE;
+
+  // Unified mux interface starts here
+  // Updates the list of resource names watched by the given watch. If an added name is new across
+  // the whole subscription, or if a removed name has no other watch interested in it, then the
+  // subscription will enqueue and attempt to send an appropriate discovery request.
+  virtual void updateWatch(const std::string& type_url, Watch* watch,
+                           const absl::flat_hash_set<std::string>& resources,
+                           const SubscriptionOptions& options) PURE;
+
+  /**
+   * Cleanup of a Watch* added by addOrUpdateWatch(). Receiving a Watch* from addOrUpdateWatch()
+   * makes you responsible for eventually invoking this cleanup.
+   * @param type_url type URL corresponding to xDS API e.g. type.googleapis.com/envoy.api.v2.Cluster
+   * @param watch the watch to be cleaned up.
+   */
+  virtual void removeWatch(const std::string& type_url, Watch* watch) PURE;
+
+  virtual bool isUnified() const { return false; }
 };
 
 using GrpcMuxPtr = std::unique_ptr<GrpcMux>;

source/common/config/grpc_mux_impl.cc

@@ -381,5 +381,15 @@ void GrpcMuxImpl::drainRequests() {
   grpc_stream_.maybeUpdateQueueSizeStat(request_queue_->size());
 }
 
+void NullGrpcMuxImpl::updateWatch(const std::string&, Watch*,
+                                  const absl::flat_hash_set<std::string>&,
+                                  const SubscriptionOptions&) {
+  throw EnvoyException("ADS must be configured to support an ADS config source");
+}
+
+void NullGrpcMuxImpl::removeWatch(const std::string&, Watch*) {
+  throw EnvoyException("ADS must be configured to support an ADS config source");
+}
+
 } // namespace Config
 } // namespace Envoy

source/common/config/grpc_mux_impl.h

@@ -88,6 +88,12 @@ class GrpcMuxImpl : public GrpcMux,
   void setRetryTimer();
   void sendDiscoveryRequest(const std::string& type_url);
 
+  void removeWatch(const std::string&, Watch*) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
+  void updateWatch(const std::string&, Watch*, const absl::flat_hash_set<std::string>&,
+                   const SubscriptionOptions&) override {
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+  }
+
   struct GrpcMuxWatchImpl : public GrpcMuxWatch {
     GrpcMuxWatchImpl(const absl::flat_hash_set<std::string>& resources,
                      SubscriptionCallbacks& callbacks, OpaqueResourceDecoder& resource_decoder,
@@ -216,6 +222,9 @@ class NullGrpcMuxImpl : public GrpcMux,
     ExceptionUtil::throwEnvoyException("ADS must be configured to support an ADS config source");
   }
 
+  void updateWatch(const std::string&, Watch*, const absl::flat_hash_set<std::string>&,
+                   const SubscriptionOptions&) override;
+  void removeWatch(const std::string&, Watch*) override;
   void requestOnDemandUpdate(const std::string&, const absl::flat_hash_set<std::string>&) override {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
   }

source/common/config/new_grpc_mux_impl.h

@@ -131,14 +131,14 @@ class NewGrpcMuxImpl
     const SubscriptionOptions options_;
   };
 
-  void removeWatch(const std::string& type_url, Watch* watch);
+  void removeWatch(const std::string& type_url, Watch* watch) override;
 
   // Updates the list of resource names watched by the given watch. If an added name is new across
   // the whole subscription, or if a removed name has no other watch interested in it, then the
   // subscription will enqueue and attempt to send an appropriate discovery request.
   void updateWatch(const std::string& type_url, Watch* watch,
                    const absl::flat_hash_set<std::string>& resources,
-                   const SubscriptionOptions& options);
+                   const SubscriptionOptions& options) override;
 
   // Adds a subscription for the type_url to the subscriptions map and order list.
   void addSubscription(const std::string& type_url, bool use_namespace_matching,

source/common/config/xds_mux/BUILD

@@ -51,3 +51,24 @@ envoy_cc_library(
         "@envoy_api//envoy/service/discovery/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_library(
+    name = "grpc_mux_lib",
+    srcs = ["grpc_mux_impl.cc"],
+    hdrs = ["grpc_mux_impl.h"],
+    deps = [
+        ":delta_subscription_state_lib",
+        ":sotw_subscription_state_lib",
+        "//envoy/event:dispatcher_interface",
+        "//envoy/grpc:async_client_interface",
+        "//source/common/config:api_version_lib",
+        "//source/common/config:decoded_resource_lib",
+        "//source/common/config:grpc_stream_lib",
+        "//source/common/config:pausable_ack_queue_lib",
+        "//source/common/config:watch_map_lib",
+        "//source/common/config:xds_context_params_lib",
+        "//source/common/config:xds_resource_lib",
+        "//source/common/memory:utils_lib",
+        "@envoy_api//envoy/service/discovery/v3:pkg_cc_proto",
+    ],
+)

source/common/config/xds_mux/delta_subscription_state.h

@@ -93,6 +93,20 @@ class DeltaSubscriptionState
   std::set<std::string> names_removed_;
 };
 
+class DeltaSubscriptionStateFactory : public SubscriptionStateFactory<DeltaSubscriptionState> {
+public:
+  DeltaSubscriptionStateFactory(Event::Dispatcher& dispatcher) : dispatcher_(dispatcher) {}
+  ~DeltaSubscriptionStateFactory() override = default;
+  std::unique_ptr<DeltaSubscriptionState>
+  makeSubscriptionState(const std::string& type_url, UntypedConfigUpdateCallbacks& callbacks,
+                        OpaqueResourceDecoder&, const bool wildcard) override {
+    return std::make_unique<DeltaSubscriptionState>(type_url, callbacks, dispatcher_, wildcard);
+  }
+
+private:
+  Event::Dispatcher& dispatcher_;
+};
+
 } // namespace XdsMux
 } // namespace Config
 } // namespace Envoy