[Filter Fusion] Create method and Fused Filter TypeName Implementation

PiperOrigin-RevId: 749157291

Author: Vignesh2208

src/core/BUILD

@@ -9178,12 +9178,17 @@ grpc_cc_library(
     ],
     external_deps = [
         "absl/status",
+        "absl/strings",
+        "absl/log:check",
         "absl/log",
+        "absl/memory",
     ],
     deps = [
         "call_filters",
         "call_final_info",
+        "channel_args",
         "metadata",
+        "status_helper",
         "type_list",
         "//:grpc_base",
         "//:grpc_public_hdrs",

src/core/call/filter_fusion.h

@@ -16,15 +16,25 @@
 #define GRPC_SRC_CORE_CALL_FILTER_FUSION_H
 #include <grpc/impl/grpc_types.h>
 
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <tuple>
 #include <type_traits>
 #include <utility>
 
+#include "absl/log/check.h"
 #include "absl/log/log.h"
+#include "absl/memory/memory.h"
 #include "absl/status/status.h"
+#include "absl/strings/str_join.h"
 #include "src/core/call/call_filters.h"
 #include "src/core/call/metadata.h"
+#include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/promise_based_filter.h"
 #include "src/core/lib/transport/call_final_info.h"
+#include "src/core/lib/transport/transport.h"
+#include "src/core/util/status_helper.h"
 #include "src/core/util/type_list.h"
 
 struct grpc_transport_op;
@@ -893,6 +903,58 @@ GRPC_FUSE_METHOD(OnClientToServerMessage, MessageHandle, true);
 GRPC_FUSE_METHOD(OnServerToClientMessage, MessageHandle, false);
 GRPC_FUSE_METHOD(OnFinalize, const grpc_call_final_info*, true);
 
+template <typename... Filters>
+struct FilterWrapper;
+
+template <typename Filter>
+struct FilterWrapper<Filter> {
+  FilterWrapper(const ChannelArgs& args, ChannelFilter::Args filter_args)
+      : filter_(Filter::Create(args, filter_args)) {}
+
+  absl::Status status() const { return filter_.status(); }
+  bool StartTransportOp(grpc_transport_op* op) {
+    CHECK(filter_.ok());
+    return (*filter_)->StartTransportOp(op);
+  }
+
+  bool GetChannelInfo(const grpc_channel_info* info) {
+    CHECK(filter_.ok());
+    return (*filter_)->GetChannelInfo(info);
+  }
+
+ private:
+  absl::StatusOr<std::unique_ptr<Filter>> filter_;
+};
+
+template <typename Filter0, typename... Filters>
+struct FilterWrapper<Filter0, Filters...> : public FilterWrapper<Filters...> {
+  FilterWrapper(const ChannelArgs& args, ChannelFilter::Args filter_args)
+      : filter0_(Filter0::Create(args, filter_args)),
+        FilterWrapper<Filters...>(args, filter_args) {}
+
+  absl::Status status() const {
+    if (filter0_.ok()) {
+      return FilterWrapper<Filters...>::status();
+    }
+    return filter0_.status();
+  }
+
+  bool StartTransportOp(grpc_transport_op* op) {
+    CHECK(filter0_.ok());
+    return (*filter0_)->StartTransportOp(op) ||
+           FilterWrapper<Filters...>::StartTransportOp(op);
+  }
+
+  bool GetChannelInfo(const grpc_channel_info* info) {
+    CHECK(filter0_.ok());
+    return (*filter0_)->GetChannelInfo(info) ||
+           FilterWrapper<Filters...>::GetChannelInfo(info);
+  }
+
+ private:
+  absl::StatusOr<std::unique_ptr<Filter0>> filter0_;
+};
+
 #undef GRPC_FUSE_METHOD
 
 template <typename... Filters>
@@ -902,19 +964,29 @@ class FusedFilter : public ImplementChannelFilter<FusedFilter<Filters...>>,
   static const grpc_channel_filter kFilter;
 
   static absl::string_view TypeName() {
-    // TODO(vigneshbabu): - Concatenate the names of the constituent filters.
-    return "fused_filter";
+    static const std::string kName = absl::StrCat(
+        "Fused_Filter_",
+        absl::StrJoin(std::make_tuple(Filters::TypeName()...), "_"));
+    return kName;
   }
 
   static absl::StatusOr<std::unique_ptr<FusedFilter<Filters...>>> Create(
-      const ChannelArgs& args, ChannelFilter::Args) {
-    // TODO(vigneshbabu): - Implement this.
-    LOG(FATAL) << "Not implemented";
-    return nullptr;
+      const ChannelArgs& args, ChannelFilter::Args filter_args) {
+    auto filters_wrapper =
+        std::make_unique<FilterWrapper<Filters...>>(args, filter_args);
+    GRPC_RETURN_IF_ERROR(filters_wrapper->status());
+    return absl::WrapUnique<FusedFilter<Filters...>>(
+        new FusedFilter<Filters...>(std::move(filters_wrapper)));
   }
 
   static constexpr bool IsFused = true;
 
+  static constexpr bool FusedFilterHasAsyncErrorInterceptor() {
+    return (
+        promise_filter_detail::CallHasAsyncErrorInterceptor<Filters>::value ||
+        ...);
+  }
+
   class Call : public FuseOnClientInitialMetadata<FusedFilter, Filters...>,
                public FuseOnServerInitialMetadata<FusedFilter, Filters...>,
                public FuseOnClientToServerMessage<FusedFilter, Filters...>,
@@ -947,27 +1019,18 @@ class FusedFilter : public ImplementChannelFilter<FusedFilter<Filters...>>,
   };
 
   bool StartTransportOp(grpc_transport_op* op) override {
-    return StartTransportOpInternal(op, Typelist<Filters...>());
+    return filters_->StartTransportOp(op);
   }
 
   bool GetChannelInfo(const grpc_channel_info* info) override {
-    return GetChannelInfoInternal(info, Typelist<Filters...>());
+    return filters_->GetChannelInfo(info);
   }
 
  private:
-  template <typename... FilterTypes>
-  bool StartTransportOpInternal(grpc_transport_op* op,
-                                Typelist<FilterTypes...>) {
-    return (std::get<FilterTypes>(filters_).StartTransportOp(op) || ...);
-  }
-
-  template <typename... FilterTypes>
-  bool GetChannelInfoInternal(const grpc_channel_info* info,
-                              Typelist<FilterTypes...>) {
-    return (std::get<FilterTypes>(filters_).GetChannelInfo(info) || ...);
-  }
+  explicit FusedFilter(std::unique_ptr<FilterWrapper<Filters...>> filters)
+      : filters_(std::move(filters)) {};
 
-  std::tuple<Filters...> filters_;
+  std::unique_ptr<FilterWrapper<Filters...>> filters_;
 };
 
 }  // namespace filters_detail

src/core/lib/channel/promise_based_filter.h

@@ -167,14 +167,37 @@ inline constexpr bool HasAnyAsyncErrorInterceptor(Interceptors...) {
   return (HasAsyncErrorInterceptor<Interceptors>::value || ...);
 }
 
+// value is true if Derived has a member called IsFused.
+template <typename Derived>
+struct IsFusedFilter {
+  template <typename V>
+  static std::true_type test(decltype(&V::IsFused));  // SFINAE context
+  template <typename V>
+  static std::false_type test(...);
+
+  using type = decltype(test<Derived>(nullptr));
+  static constexpr bool value = std::is_same_v<type, std::true_type>;
+};
+
+template <typename Derived, typename Ignored = void>
+struct CallHasAsyncErrorInterceptor;
+
 // Composite for a given channel type to determine if any of its interceptors
 // fall into this category: later code should use this.
 template <typename Derived>
-inline constexpr bool CallHasAsyncErrorInterceptor() {
-  return HasAnyAsyncErrorInterceptor(&Derived::Call::OnClientToServerMessage,
-                                     &Derived::Call::OnServerInitialMetadata,
-                                     &Derived::Call::OnServerToClientMessage);
-}
+struct CallHasAsyncErrorInterceptor<
+    Derived, std::enable_if_t<!IsFusedFilter<Derived>::value>> {
+  static constexpr bool value =
+      HasAnyAsyncErrorInterceptor(&Derived::Call::OnClientToServerMessage,
+                                  &Derived::Call::OnServerInitialMetadata,
+                                  &Derived::Call::OnServerToClientMessage);
+};
+
+template <typename Derived>
+struct CallHasAsyncErrorInterceptor<
+    Derived, std::enable_if_t<IsFusedFilter<Derived>::value>> {
+  static constexpr bool value = Derived::FusedFilterHasAsyncErrorInterceptor();
+};
 
 // Given a boolean X export a type:
 // either T if X is true
@@ -246,7 +269,7 @@ struct FilterCallData {
   GPR_NO_UNIQUE_ADDRESS CallWrapper<Derived> call;
   GPR_NO_UNIQUE_ADDRESS
   typename TypeIfNeeded<Latch<ServerMetadataHandle>,
-                        CallHasAsyncErrorInterceptor<Derived>()>::Type
+                        CallHasAsyncErrorInterceptor<Derived>::value>::Type
       error_latch;
   GPR_NO_UNIQUE_ADDRESS
   typename TypeIfNeeded<
@@ -317,7 +340,7 @@ auto MapResult(void (Derived::Call::*fn)(ServerMetadata&, Derived*), Promise x,
 // For fused filters whose OnServerTrailingMetadata takes pointer to the
 // channel.
 template <typename P, typename Call, typename Derived,
-          typename = std::enable_if_t<Derived::IsFused>>
+          typename = std::enable_if_t<IsFusedFilter<Derived>::value>>
 auto MapResult(void (Call::*fn)(ServerMetadata&, Derived*), P x,
                FilterCallData<Derived>* call_data) {
   DCHECK(fn == &Derived::Call::OnServerTrailingMetadata);
@@ -338,7 +361,7 @@ auto MapResult(void (Call::*fn)(ServerMetadata&, Derived*), P x,
 // For fused filters whose OnServerTrailingMetadata does not take pointer to the
 // channel.
 template <typename P, typename Call, typename Derived,
-          typename = std::enable_if_t<Derived::IsFused>>
+          typename = std::enable_if_t<IsFusedFilter<Derived>::value>>
 auto MapResult(void (Call::*fn)(ServerMetadata&), P x,
                FilterCallData<Derived>* call_data) {
   DCHECK(fn == &Derived::Call::OnServerTrailingMetadata);
@@ -1183,7 +1206,8 @@ class ImplementChannelFilter : public ChannelFilter,
     return promise_filter_detail::MapResult(
         &Derived::Call::OnServerTrailingMetadata,
         promise_filter_detail::RaceAsyncCompletion<
-            promise_filter_detail::CallHasAsyncErrorInterceptor<Derived>()>::
+            promise_filter_detail::CallHasAsyncErrorInterceptor<
+                Derived>::value>::
             Run(promise_filter_detail::RunCall(
                     &Derived::Call::OnClientInitialMetadata,
                     std::move(call_args), std::move(next_promise_factory),