[promises] Work towards making the v3 filter interface available
The code in promise_based_filter allows http_client_filter to use the v3 APIs for filters. The implementation of the mapping from v3 -> v2.1 is as yet incomplete, but sufficient for this sample.
The long term aim will be to convert all filters to this API, and then transition the runtime from the current promise based stack to a new stack that uses these interception points natively.
PiperOrigin-RevId: 586188548
diff --git a/src/core/ext/filters/http/client/http_client_filter.cc b/src/core/ext/filters/http/client/http_client_filter.cc
index 8b547ba..07fcba6 100644
--- a/src/core/ext/filters/http/client/http_client_filter.cc
+++ b/src/core/ext/filters/http/client/http_client_filter.cc
@@ -51,6 +51,9 @@
namespace grpc_core {
+const NoInterceptor HttpClientFilter::Call::OnServerToClientMessage;
+const NoInterceptor HttpClientFilter::Call::OnClientToServerMessage;
+
const grpc_channel_filter HttpClientFilter::kFilter =
MakePromiseBasedFilter<HttpClientFilter, FilterEndpoint::kClient,
kFilterExaminesServerInitialMetadata>("http-client");
@@ -105,40 +108,27 @@
}
} // namespace
-ArenaPromise<ServerMetadataHandle> HttpClientFilter::MakeCallPromise(
- CallArgs call_args, NextPromiseFactory next_promise_factory) {
- auto& md = call_args.client_initial_metadata;
- if (test_only_use_put_requests_) {
- md->Set(HttpMethodMetadata(), HttpMethodMetadata::kPut);
+void HttpClientFilter::Call::OnClientInitialMetadata(ClientMetadata& md,
+ HttpClientFilter* filter) {
+ if (filter->test_only_use_put_requests_) {
+ md.Set(HttpMethodMetadata(), HttpMethodMetadata::kPut);
} else {
- md->Set(HttpMethodMetadata(), HttpMethodMetadata::kPost);
+ md.Set(HttpMethodMetadata(), HttpMethodMetadata::kPost);
}
- md->Set(HttpSchemeMetadata(), scheme_);
- md->Set(TeMetadata(), TeMetadata::kTrailers);
- md->Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
- md->Set(UserAgentMetadata(), user_agent_.Ref());
+ md.Set(HttpSchemeMetadata(), filter->scheme_);
+ md.Set(TeMetadata(), TeMetadata::kTrailers);
+ md.Set(ContentTypeMetadata(), ContentTypeMetadata::kApplicationGrpc);
+ md.Set(UserAgentMetadata(), filter->user_agent_.Ref());
+}
- auto* initial_metadata_err =
- GetContext<Arena>()->New<Latch<ServerMetadataHandle>>();
+absl::Status HttpClientFilter::Call::OnServerInitialMetadata(
+ ServerMetadata& md) {
+ return CheckServerMetadata(&md);
+}
- call_args.server_initial_metadata->InterceptAndMap(
- [initial_metadata_err](
- ServerMetadataHandle md) -> absl::optional<ServerMetadataHandle> {
- auto r = CheckServerMetadata(md.get());
- if (!r.ok()) {
- initial_metadata_err->Set(ServerMetadataFromStatus(r));
- return absl::nullopt;
- }
- return std::move(md);
- });
-
- return Race(initial_metadata_err->Wait(),
- Map(next_promise_factory(std::move(call_args)),
- [](ServerMetadataHandle md) -> ServerMetadataHandle {
- auto r = CheckServerMetadata(md.get());
- if (!r.ok()) return ServerMetadataFromStatus(r);
- return md;
- }));
+absl::Status HttpClientFilter::Call::OnServerTrailingMetadata(
+ ServerMetadata& md) {
+ return CheckServerMetadata(&md);
}
HttpClientFilter::HttpClientFilter(HttpSchemeMetadata::ValueType scheme,
diff --git a/src/core/ext/filters/http/client/http_client_filter.h b/src/core/ext/filters/http/client/http_client_filter.h
index 298daf0..f66fc32 100644
--- a/src/core/ext/filters/http/client/http_client_filter.h
+++ b/src/core/ext/filters/http/client/http_client_filter.h
@@ -20,28 +20,31 @@
#include <grpc/support/port_platform.h>
-#include "absl/status/statusor.h"
-
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/channel_fwd.h"
#include "src/core/lib/channel/promise_based_filter.h"
-#include "src/core/lib/promise/arena_promise.h"
#include "src/core/lib/slice/slice.h"
#include "src/core/lib/transport/metadata_batch.h"
#include "src/core/lib/transport/transport.h"
+#include "absl/status/statusor.h"
namespace grpc_core {
-class HttpClientFilter : public ChannelFilter {
+class HttpClientFilter : public ImplementChannelFilter<HttpClientFilter> {
public:
static const grpc_channel_filter kFilter;
static absl::StatusOr<HttpClientFilter> Create(
const ChannelArgs& args, ChannelFilter::Args filter_args);
- // Construct a promise for one call.
- ArenaPromise<ServerMetadataHandle> MakeCallPromise(
- CallArgs call_args, NextPromiseFactory next_promise_factory) override;
+ class Call {
+ public:
+ void OnClientInitialMetadata(ClientMetadata& md, HttpClientFilter* filter);
+ absl::Status OnServerInitialMetadata(ServerMetadata& md);
+ absl::Status OnServerTrailingMetadata(ServerMetadata& md);
+ static const NoInterceptor OnClientToServerMessage;
+ static const NoInterceptor OnServerToClientMessage;
+ };
private:
HttpClientFilter(HttpSchemeMetadata::ValueType scheme, Slice user_agent,
diff --git a/src/core/lib/channel/promise_based_filter.h b/src/core/lib/channel/promise_based_filter.h
index 19efe50..bfddbd2 100644
--- a/src/core/lib/channel/promise_based_filter.h
+++ b/src/core/lib/channel/promise_based_filter.h
@@ -19,8 +19,10 @@
// promise-style. Most of this will be removed once the promises conversion is
// completed.
+#include <grpc/event_engine/event_engine.h>
+#include <grpc/grpc.h>
+#include <grpc/support/log.h>
#include <grpc/support/port_platform.h>
-
#include <stdint.h>
#include <stdlib.h>
@@ -31,18 +33,8 @@
#include <type_traits>
#include <utility>
-#include "absl/container/inlined_vector.h"
-#include "absl/functional/function_ref.h"
-#include "absl/meta/type_traits.h"
-#include "absl/status/status.h"
-#include "absl/strings/string_view.h"
-#include "absl/types/optional.h"
-
-#include <grpc/event_engine/event_engine.h>
-#include <grpc/grpc.h>
-#include <grpc/support/log.h>
-
#include "src/core/lib/channel/call_finalization.h"
+#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/channel_fwd.h"
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/channel/context.h"
@@ -66,6 +58,12 @@
#include "src/core/lib/transport/error_utils.h"
#include "src/core/lib/transport/metadata_batch.h"
#include "src/core/lib/transport/transport.h"
+#include "absl/container/inlined_vector.h"
+#include "absl/functional/function_ref.h"
+#include "absl/meta/type_traits.h"
+#include "absl/status/status.h"
+#include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
namespace grpc_core {
@@ -122,6 +120,223 @@
grpc_event_engine::experimental::GetDefaultEventEngine();
};
+struct NoInterceptor {};
+
+namespace promise_filter_detail {
+
+inline constexpr bool HasAsyncErrorInterceptor() { return false; }
+
+inline constexpr bool HasAsyncErrorInterceptor(const NoInterceptor*) {
+ return false;
+}
+
+template <typename T, typename... A>
+inline constexpr bool HasAsyncErrorInterceptor(absl::Status (T::*)(A...)) {
+ return true;
+}
+
+template <typename T, typename... A>
+inline constexpr bool HasAsyncErrorInterceptor(void (T::*)(A...)) {
+ return false;
+}
+
+template <typename I1, typename I2, typename... Interceptors>
+inline constexpr bool HasAsyncErrorInterceptor(I1 i1, I2 i2,
+ Interceptors... interceptors) {
+ return HasAsyncErrorInterceptor(i1) || HasAsyncErrorInterceptor(i2) ||
+ HasAsyncErrorInterceptor(interceptors...);
+}
+
+inline constexpr bool HasChannelAccess() { return false; }
+
+inline constexpr bool HasChannelAccess(const NoInterceptor*) { return false; }
+
+template <typename T, typename R, typename A>
+inline constexpr bool HasChannelAccess(R (T::*)(A)) {
+ return false;
+}
+
+template <typename T, typename R, typename A, typename C>
+inline constexpr bool HasChannelAccess(R (T::*)(A, C)) {
+ return true;
+}
+
+template <typename I1, typename I2, typename... Interceptors>
+inline constexpr bool HasChannelAccess(I1 i1, I2 i2,
+ Interceptors... interceptors) {
+ return HasChannelAccess(i1) || HasChannelAccess(i2) ||
+ HasChannelAccess(interceptors...);
+}
+
+template <typename Derived>
+inline constexpr bool CallHasAsyncErrorInterceptor() {
+ return HasAsyncErrorInterceptor(&Derived::Call::OnClientToServerMessage,
+ &Derived::Call::OnServerInitialMetadata,
+ &Derived::Call::OnServerToClientMessage);
+}
+
+template <typename Derived>
+inline constexpr bool CallHasChannelAccess() {
+ return HasAsyncErrorInterceptor(&Derived::Call::OnClientInitialMetadata,
+ &Derived::Call::OnClientToServerMessage,
+ &Derived::Call::OnServerInitialMetadata,
+ &Derived::Call::OnServerToClientMessage,
+ &Derived::Call::OnServerTrailingMetadata);
+}
+
+template <typename T, bool X>
+struct TypeIfNeeded;
+
+template <typename T>
+struct TypeIfNeeded<T, false> {
+ struct Type {
+ Type() = default;
+ template <typename Whatever>
+ explicit Type(Whatever) : Type() {}
+ };
+};
+
+template <typename T>
+struct TypeIfNeeded<T, true> {
+ using Type = T;
+};
+
+template <bool X>
+struct RaceAsyncCompletion;
+template <>
+struct RaceAsyncCompletion<false> {
+ template <typename Promise>
+ static Promise Run(Promise x, void*) {
+ return x;
+ }
+};
+
+template <>
+struct RaceAsyncCompletion<true> {
+ template <typename Promise>
+ static Promise Run(Promise x, Latch<ServerMetadataHandle>* latch) {
+ return Race(latch->Wait(), std::move(x));
+ }
+};
+
+template <typename Derived>
+struct FilterCallData {
+ FilterCallData(Derived* channel) : channel(channel) {}
+ GPR_NO_UNIQUE_ADDRESS typename Derived::Call call;
+ GPR_NO_UNIQUE_ADDRESS
+ typename TypeIfNeeded<Latch<ServerMetadataHandle>,
+ CallHasAsyncErrorInterceptor<Derived>()>::Type
+ error_latch;
+ GPR_NO_UNIQUE_ADDRESS
+ typename TypeIfNeeded<Derived*, CallHasChannelAccess<Derived>()>::Type
+ channel;
+};
+
+template <typename Promise>
+auto MapResult(const NoInterceptor*, Promise x, void*) {
+ return x;
+}
+
+template <typename Promise, typename Derived>
+auto MapResult(absl::Status (Derived::Call::*fn)(ServerMetadata&), Promise x,
+ FilterCallData<Derived>* call_data) {
+ return Map(std::move(x), [call_data, fn](ServerMetadataHandle md) {
+ auto status = (call_data->call.*fn)(*md);
+ if (!status.ok()) return ServerMetadataFromStatus(status);
+ return md;
+ });
+}
+
+inline auto RunCall(const NoInterceptor*, CallArgs call_args,
+ NextPromiseFactory next_promise_factory, void*) {
+ return next_promise_factory(std::move(call_args));
+}
+
+template <typename Derived>
+inline auto RunCall(void (Derived::Call::*fn)(ClientMetadata& md),
+ CallArgs call_args, NextPromiseFactory next_promise_factory,
+ FilterCallData<Derived>* call_data) {
+ (call_data->call.*fn)(*call_args.client_initial_metadata);
+ return next_promise_factory(std::move(call_args));
+}
+
+template <typename Derived>
+inline auto RunCall(void (Derived::Call::*fn)(ClientMetadata& md,
+ Derived* channel),
+ CallArgs call_args, NextPromiseFactory next_promise_factory,
+ FilterCallData<Derived>* call_data) {
+ (call_data->call.*fn)(*call_args.client_initial_metadata, call_data->channel);
+ return next_promise_factory(std::move(call_args));
+}
+
+inline void InterceptClientToServerMessage(const NoInterceptor*, void*,
+ CallArgs&) {}
+
+inline void InterceptServerInitialMetadata(const NoInterceptor*, void*,
+ CallArgs&) {}
+
+template <typename Derived>
+inline void InterceptServerInitialMetadata(
+ absl::Status (Derived::Call::*fn)(ServerMetadata&),
+ FilterCallData<Derived>* call_data, CallArgs& call_args) {
+ call_args.server_initial_metadata->InterceptAndMap(
+ [call_data,
+ fn](ServerMetadataHandle md) -> absl::optional<ServerMetadataHandle> {
+ auto status = (call_data->call.*fn)(*md);
+ if (!status.ok() && !call_data->error_latch.is_set()) {
+ call_data->error_latch.Set(ServerMetadataFromStatus(status));
+ return absl::nullopt;
+ }
+ return md;
+ });
+}
+
+inline void InterceptServerToClientMessage(const NoInterceptor*, void*,
+ CallArgs&) {}
+
+template <typename Derived>
+absl::enable_if_t<std::is_empty<FilterCallData<Derived>>::value,
+ FilterCallData<Derived>*>
+MakeFilterCall(Derived*) {
+ static FilterCallData<Derived> call{nullptr};
+ return &call;
+}
+
+template <typename Derived>
+absl::enable_if_t<!std::is_empty<FilterCallData<Derived>>::value,
+ FilterCallData<Derived>*>
+MakeFilterCall(Derived* derived) {
+ return GetContext<Arena>()->ManagedNew<FilterCallData<Derived>>(derived);
+}
+
+} // namespace promise_filter_detail
+
+template <typename Derived>
+class ImplementChannelFilter : public ChannelFilter {
+ public:
+ ArenaPromise<ServerMetadataHandle> MakeCallPromise(
+ CallArgs call_args, NextPromiseFactory next_promise_factory) final {
+ auto* call = promise_filter_detail::MakeFilterCall<Derived>(
+ static_cast<Derived*>(this));
+ promise_filter_detail::InterceptClientToServerMessage(
+ &Derived::Call::OnClientToServerMessage, call, call_args);
+ promise_filter_detail::InterceptServerInitialMetadata(
+ &Derived::Call::OnServerInitialMetadata, call, call_args);
+ promise_filter_detail::InterceptServerToClientMessage(
+ &Derived::Call::OnServerToClientMessage, call, call_args);
+ return promise_filter_detail::MapResult(
+ &Derived::Call::OnServerTrailingMetadata,
+ promise_filter_detail::RaceAsyncCompletion<
+ promise_filter_detail::CallHasAsyncErrorInterceptor<Derived>()>::
+ Run(promise_filter_detail::RunCall(
+ &Derived::Call::OnClientInitialMetadata,
+ std::move(call_args), std::move(next_promise_factory),
+ call),
+ &call->error_latch),
+ call);
+ }
+};
+
// Designator for whether a filter is client side or server side.
// Please don't use this outside calls to MakePromiseBasedFilter - it's
// intended to be deleted once the promise conversion is complete.
