|
|
|
|
@ -534,13 +534,14 @@ inline void InterceptClientToServerMessage(const NoInterceptor*, void*, void*, |
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptClientToServerMessage( |
|
|
|
|
ServerMetadataHandle (Derived::Call::*fn)(const Message&), |
|
|
|
|
typename Derived::Call* call, Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
typename Derived::Call* call, Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientToServerMessage); |
|
|
|
|
call_spine->client_to_server_messages().receiver.InterceptAndMap( |
|
|
|
|
[call, call_spine](MessageHandle msg) -> absl::optional<MessageHandle> { |
|
|
|
|
auto return_md = call->OnClientToServerMessage(*msg); |
|
|
|
|
if (return_md == nullptr) return std::move(msg); |
|
|
|
|
return call_spine->Cancel(std::move(return_md)); |
|
|
|
|
call_spine->PushServerTrailingMetadata(std::move(return_md)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -548,14 +549,15 @@ template <typename Derived> |
|
|
|
|
inline void InterceptClientToServerMessage( |
|
|
|
|
ServerMetadataHandle (Derived::Call::*fn)(const Message&, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientToServerMessage); |
|
|
|
|
call_spine->client_to_server_messages().receiver.InterceptAndMap( |
|
|
|
|
[call, call_spine, |
|
|
|
|
channel](MessageHandle msg) -> absl::optional<MessageHandle> { |
|
|
|
|
auto return_md = call->OnClientToServerMessage(*msg, channel); |
|
|
|
|
if (return_md == nullptr) return std::move(msg); |
|
|
|
|
return call_spine->Cancel(std::move(return_md)); |
|
|
|
|
call_spine->PushServerTrailingMetadata(std::move(return_md)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -563,7 +565,7 @@ template <typename Derived> |
|
|
|
|
inline void InterceptClientToServerMessage( |
|
|
|
|
MessageHandle (Derived::Call::*fn)(MessageHandle, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientToServerMessage); |
|
|
|
|
call_spine->client_to_server_messages().receiver.InterceptAndMap( |
|
|
|
|
[call, channel](MessageHandle msg) { |
|
|
|
|
@ -575,24 +577,26 @@ template <typename Derived> |
|
|
|
|
inline void InterceptClientToServerMessage( |
|
|
|
|
absl::StatusOr<MessageHandle> (Derived::Call::*fn)(MessageHandle, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientToServerMessage); |
|
|
|
|
call_spine->client_to_server_messages().receiver.InterceptAndMap( |
|
|
|
|
[call, call_spine, |
|
|
|
|
channel](MessageHandle msg) -> absl::optional<MessageHandle> { |
|
|
|
|
auto r = call->OnClientToServerMessage(std::move(msg), channel); |
|
|
|
|
if (r.ok()) return std::move(*r); |
|
|
|
|
return call_spine->Cancel(ServerMetadataFromStatus(r.status())); |
|
|
|
|
call_spine->PushServerTrailingMetadata( |
|
|
|
|
ServerMetadataFromStatus(r.status())); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
inline void InterceptClientInitialMetadata(const NoInterceptor*, void*, void*, |
|
|
|
|
CallSpineInterface*) {} |
|
|
|
|
PipeBasedCallSpine*) {} |
|
|
|
|
|
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptClientInitialMetadata( |
|
|
|
|
void (Derived::Call::*fn)(ClientMetadata& md), typename Derived::Call* call, |
|
|
|
|
Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call](ClientMetadataHandle md) { |
|
|
|
|
@ -605,7 +609,7 @@ template <typename Derived> |
|
|
|
|
inline void InterceptClientInitialMetadata( |
|
|
|
|
void (Derived::Call::*fn)(ClientMetadata& md, Derived* channel), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call, channel](ClientMetadataHandle md) { |
|
|
|
|
@ -617,14 +621,15 @@ inline void InterceptClientInitialMetadata( |
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptClientInitialMetadata( |
|
|
|
|
ServerMetadataHandle (Derived::Call::*fn)(ClientMetadata& md), |
|
|
|
|
typename Derived::Call* call, Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
typename Derived::Call* call, Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call_spine, |
|
|
|
|
call](ClientMetadataHandle md) -> absl::optional<ClientMetadataHandle> { |
|
|
|
|
auto return_md = call->OnClientInitialMetadata(*md); |
|
|
|
|
if (return_md == nullptr) return std::move(md); |
|
|
|
|
return call_spine->Cancel(std::move(return_md)); |
|
|
|
|
call_spine->PushServerTrailingMetadata(std::move(return_md)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -633,28 +638,31 @@ inline void InterceptClientInitialMetadata( |
|
|
|
|
ServerMetadataHandle (Derived::Call::*fn)(ClientMetadata& md, |
|
|
|
|
Derived* channel), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call_spine, call, channel]( |
|
|
|
|
ClientMetadataHandle md) -> absl::optional<ClientMetadataHandle> { |
|
|
|
|
auto return_md = call->OnClientInitialMetadata(*md, channel); |
|
|
|
|
if (return_md == nullptr) return std::move(md); |
|
|
|
|
return call_spine->Cancel(std::move(return_md)); |
|
|
|
|
call_spine->PushServerTrailingMetadata(std::move(return_md)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptClientInitialMetadata( |
|
|
|
|
absl::Status (Derived::Call::*fn)(ClientMetadata& md), |
|
|
|
|
typename Derived::Call* call, Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
typename Derived::Call* call, Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call_spine, |
|
|
|
|
call](ClientMetadataHandle md) -> absl::optional<ClientMetadataHandle> { |
|
|
|
|
auto status = call->OnClientInitialMetadata(*md); |
|
|
|
|
if (status.ok()) return std::move(md); |
|
|
|
|
return call_spine->Cancel(ServerMetadataFromStatus(status)); |
|
|
|
|
call_spine->PushServerTrailingMetadata( |
|
|
|
|
ServerMetadataFromStatus(status)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -662,14 +670,16 @@ template <typename Derived> |
|
|
|
|
inline void InterceptClientInitialMetadata( |
|
|
|
|
absl::Status (Derived::Call::*fn)(ClientMetadata& md, Derived* channel), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call_spine, call, channel]( |
|
|
|
|
ClientMetadataHandle md) -> absl::optional<ClientMetadataHandle> { |
|
|
|
|
auto status = call->OnClientInitialMetadata(*md, channel); |
|
|
|
|
if (status.ok()) return std::move(md); |
|
|
|
|
return call_spine->Cancel(ServerMetadataFromStatus(status)); |
|
|
|
|
call_spine->PushServerTrailingMetadata( |
|
|
|
|
ServerMetadataFromStatus(status)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -681,7 +691,7 @@ absl::void_t<decltype(StatusCast<ServerMetadataHandle>( |
|
|
|
|
InterceptClientInitialMetadata(Promise (Derived::Call::*promise_factory)( |
|
|
|
|
ClientMetadata& md, Derived* channel), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(promise_factory == &Derived::Call::OnClientInitialMetadata); |
|
|
|
|
call_spine->client_initial_metadata().receiver.InterceptAndMap( |
|
|
|
|
[call, call_spine, channel](ClientMetadataHandle md) { |
|
|
|
|
@ -691,8 +701,9 @@ InterceptClientInitialMetadata(Promise (Derived::Call::*promise_factory)( |
|
|
|
|
call_spine](PromiseResult<Promise> status) mutable |
|
|
|
|
-> absl::optional<ClientMetadataHandle> { |
|
|
|
|
if (IsStatusOk(status)) return std::move(md); |
|
|
|
|
return call_spine->Cancel( |
|
|
|
|
call_spine->PushServerTrailingMetadata( |
|
|
|
|
StatusCast<ServerMetadataHandle>(std::move(status))); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
@ -766,7 +777,7 @@ inline void InterceptServerInitialMetadata(const NoInterceptor*, void*, void*, |
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptServerInitialMetadata( |
|
|
|
|
void (Derived::Call::*fn)(ServerMetadata&), typename Derived::Call* call, |
|
|
|
|
Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerInitialMetadata); |
|
|
|
|
call_spine->server_initial_metadata().sender.InterceptAndMap( |
|
|
|
|
[call](ServerMetadataHandle md) { |
|
|
|
|
@ -778,14 +789,16 @@ inline void InterceptServerInitialMetadata( |
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptServerInitialMetadata( |
|
|
|
|
absl::Status (Derived::Call::*fn)(ServerMetadata&), |
|
|
|
|
typename Derived::Call* call, Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
typename Derived::Call* call, Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerInitialMetadata); |
|
|
|
|
call_spine->server_initial_metadata().sender.InterceptAndMap( |
|
|
|
|
[call, call_spine]( |
|
|
|
|
ServerMetadataHandle md) -> absl::optional<ServerMetadataHandle> { |
|
|
|
|
auto status = call->OnServerInitialMetadata(*md); |
|
|
|
|
if (status.ok()) return std::move(md); |
|
|
|
|
return call_spine->Cancel(ServerMetadataFromStatus(status)); |
|
|
|
|
call_spine->PushServerTrailingMetadata( |
|
|
|
|
ServerMetadataFromStatus(status)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -793,7 +806,7 @@ template <typename Derived> |
|
|
|
|
inline void InterceptServerInitialMetadata( |
|
|
|
|
void (Derived::Call::*fn)(ServerMetadata&, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerInitialMetadata); |
|
|
|
|
call_spine->server_initial_metadata().sender.InterceptAndMap( |
|
|
|
|
[call, channel](ServerMetadataHandle md) { |
|
|
|
|
@ -806,14 +819,16 @@ template <typename Derived> |
|
|
|
|
inline void InterceptServerInitialMetadata( |
|
|
|
|
absl::Status (Derived::Call::*fn)(ServerMetadata&, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerInitialMetadata); |
|
|
|
|
call_spine->server_initial_metadata().sender.InterceptAndMap( |
|
|
|
|
[call, call_spine, channel]( |
|
|
|
|
ServerMetadataHandle md) -> absl::optional<ServerMetadataHandle> { |
|
|
|
|
auto status = call->OnServerInitialMetadata(*md, channel); |
|
|
|
|
if (status.ok()) return std::move(md); |
|
|
|
|
return call_spine->Cancel(ServerMetadataFromStatus(status)); |
|
|
|
|
call_spine->PullServerTrailingMetadata( |
|
|
|
|
ServerMetadataFromStatus(status)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -885,13 +900,14 @@ inline void InterceptServerToClientMessage(const NoInterceptor*, void*, void*, |
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptServerToClientMessage( |
|
|
|
|
ServerMetadataHandle (Derived::Call::*fn)(const Message&), |
|
|
|
|
typename Derived::Call* call, Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
typename Derived::Call* call, Derived*, PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerToClientMessage); |
|
|
|
|
call_spine->server_to_client_messages().sender.InterceptAndMap( |
|
|
|
|
[call, call_spine](MessageHandle msg) -> absl::optional<MessageHandle> { |
|
|
|
|
auto return_md = call->OnServerToClientMessage(*msg); |
|
|
|
|
if (return_md == nullptr) return std::move(msg); |
|
|
|
|
return call_spine->Cancel(std::move(return_md)); |
|
|
|
|
call_spine->PushServerTrailingMetadata(std::move(return_md)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -899,14 +915,15 @@ template <typename Derived> |
|
|
|
|
inline void InterceptServerToClientMessage( |
|
|
|
|
ServerMetadataHandle (Derived::Call::*fn)(const Message&, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerToClientMessage); |
|
|
|
|
call_spine->server_to_client_messages().sender.InterceptAndMap( |
|
|
|
|
[call, call_spine, |
|
|
|
|
channel](MessageHandle msg) -> absl::optional<MessageHandle> { |
|
|
|
|
auto return_md = call->OnServerToClientMessage(*msg, channel); |
|
|
|
|
if (return_md == nullptr) return std::move(msg); |
|
|
|
|
return call_spine->Cancel(std::move(return_md)); |
|
|
|
|
call_spine->PushServerTrailingMetadata(std::move(return_md)); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -914,7 +931,7 @@ template <typename Derived> |
|
|
|
|
inline void InterceptServerToClientMessage( |
|
|
|
|
MessageHandle (Derived::Call::*fn)(MessageHandle, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerToClientMessage); |
|
|
|
|
call_spine->server_to_client_messages().sender.InterceptAndMap( |
|
|
|
|
[call, channel](MessageHandle msg) { |
|
|
|
|
@ -926,14 +943,16 @@ template <typename Derived> |
|
|
|
|
inline void InterceptServerToClientMessage( |
|
|
|
|
absl::StatusOr<MessageHandle> (Derived::Call::*fn)(MessageHandle, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
PipeBasedCallSpine* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerToClientMessage); |
|
|
|
|
call_spine->server_to_client_messages().sender.InterceptAndMap( |
|
|
|
|
[call, call_spine, |
|
|
|
|
channel](MessageHandle msg) -> absl::optional<MessageHandle> { |
|
|
|
|
auto r = call->OnServerToClientMessage(std::move(msg), channel); |
|
|
|
|
if (r.ok()) return std::move(*r); |
|
|
|
|
return call_spine->Cancel(ServerMetadataFromStatus(r.status())); |
|
|
|
|
call_spine->PushServerTrailingMetadata( |
|
|
|
|
ServerMetadataFromStatus(r.status())); |
|
|
|
|
return absl::nullopt; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -942,40 +961,25 @@ inline void InterceptServerTrailingMetadata(const NoInterceptor*, void*, void*, |
|
|
|
|
|
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptServerTrailingMetadata( |
|
|
|
|
void (Derived::Call::*fn)(ServerMetadata&), typename Derived::Call* call, |
|
|
|
|
Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerTrailingMetadata); |
|
|
|
|
call_spine->server_trailing_metadata().sender.InterceptAndMap( |
|
|
|
|
[call](ServerMetadataHandle md) { |
|
|
|
|
call->OnServerTrailingMetadata(*md); |
|
|
|
|
return md; |
|
|
|
|
}); |
|
|
|
|
void (Derived::Call::*)(ServerMetadata&), typename Derived::Call*, Derived*, |
|
|
|
|
PipeBasedCallSpine*) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"InterceptServerTrailingMetadata not available for call v2.5"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptServerTrailingMetadata( |
|
|
|
|
void (Derived::Call::*fn)(ServerMetadata&, Derived*), |
|
|
|
|
typename Derived::Call* call, Derived* channel, |
|
|
|
|
CallSpineInterface* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerTrailingMetadata); |
|
|
|
|
call_spine->server_trailing_metadata().sender.InterceptAndMap( |
|
|
|
|
[call, channel](ServerMetadataHandle md) { |
|
|
|
|
call->OnServerTrailingMetadata(*md, channel); |
|
|
|
|
return md; |
|
|
|
|
}); |
|
|
|
|
void (Derived::Call::*)(ServerMetadata&, Derived*), typename Derived::Call*, |
|
|
|
|
Derived*, PipeBasedCallSpine*) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"InterceptServerTrailingMetadata not available for call v2.5"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <typename Derived> |
|
|
|
|
inline void InterceptServerTrailingMetadata( |
|
|
|
|
absl::Status (Derived::Call::*fn)(ServerMetadata&), |
|
|
|
|
typename Derived::Call* call, Derived*, CallSpineInterface* call_spine) { |
|
|
|
|
GPR_DEBUG_ASSERT(fn == &Derived::Call::OnServerTrailingMetadata); |
|
|
|
|
call_spine->server_trailing_metadata().sender.InterceptAndMap( |
|
|
|
|
[call](ServerMetadataHandle md) -> absl::optional<ServerMetadataHandle> { |
|
|
|
|
auto status = call->OnServerTrailingMetadata(*md); |
|
|
|
|
if (status.ok()) return std::move(md); |
|
|
|
|
return ServerMetadataFromStatus(status); |
|
|
|
|
}); |
|
|
|
|
absl::Status (Derived::Call::*)(ServerMetadata&), typename Derived::Call*, |
|
|
|
|
Derived*, PipeBasedCallSpine*) { |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"InterceptServerTrailingMetadata not available for call v2.5"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
inline void InterceptFinalize(const NoInterceptor*, void*, void*) {} |
|
|
|
|
@ -1085,23 +1089,20 @@ class ImplementChannelFilter : public ChannelFilter, |
|
|
|
|
GetContext<Arena>() |
|
|
|
|
->ManagedNew<promise_filter_detail::CallWrapper<Derived>>( |
|
|
|
|
static_cast<Derived*>(this)); |
|
|
|
|
auto* c = DownCast<PipeBasedCallSpine*>(call_spine); |
|
|
|
|
auto* d = static_cast<Derived*>(this); |
|
|
|
|
promise_filter_detail::InterceptClientInitialMetadata( |
|
|
|
|
&Derived::Call::OnClientInitialMetadata, call, |
|
|
|
|
static_cast<Derived*>(this), call_spine); |
|
|
|
|
&Derived::Call::OnClientInitialMetadata, call, d, c); |
|
|
|
|
promise_filter_detail::InterceptClientToServerMessage( |
|
|
|
|
&Derived::Call::OnClientToServerMessage, call, |
|
|
|
|
static_cast<Derived*>(this), call_spine); |
|
|
|
|
&Derived::Call::OnClientToServerMessage, call, d, c); |
|
|
|
|
promise_filter_detail::InterceptServerInitialMetadata( |
|
|
|
|
&Derived::Call::OnServerInitialMetadata, call, |
|
|
|
|
static_cast<Derived*>(this), call_spine); |
|
|
|
|
&Derived::Call::OnServerInitialMetadata, call, d, c); |
|
|
|
|
promise_filter_detail::InterceptServerToClientMessage( |
|
|
|
|
&Derived::Call::OnServerToClientMessage, call, |
|
|
|
|
static_cast<Derived*>(this), call_spine); |
|
|
|
|
&Derived::Call::OnServerToClientMessage, call, d, c); |
|
|
|
|
promise_filter_detail::InterceptServerTrailingMetadata( |
|
|
|
|
&Derived::Call::OnServerTrailingMetadata, call, |
|
|
|
|
static_cast<Derived*>(this), call_spine); |
|
|
|
|
promise_filter_detail::InterceptFinalize(&Derived::Call::OnFinalize, |
|
|
|
|
static_cast<Derived*>(this), call); |
|
|
|
|
&Derived::Call::OnServerTrailingMetadata, call, d, c); |
|
|
|
|
promise_filter_detail::InterceptFinalize(&Derived::Call::OnFinalize, d, |
|
|
|
|
call); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Polyfill for the original promise scheme.
|
|
|
|
|
@ -1840,6 +1841,15 @@ struct BaseCallDataMethods { |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
// The type of object returned by a filter's Create method.
|
|
|
|
|
template <typename T> |
|
|
|
|
using CreatedType = typename decltype(T::Create(ChannelArgs(), {}))::value_type; |
|
|
|
|
|
|
|
|
|
template <typename GrpcChannelOrCallElement> |
|
|
|
|
inline ChannelFilter* ChannelFilterFromElem(GrpcChannelOrCallElement* elem) { |
|
|
|
|
return *static_cast<ChannelFilter**>(elem->channel_data); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
template <typename CallData, uint8_t kFlags> |
|
|
|
|
struct CallDataFilterWithFlagsMethods { |
|
|
|
|
static absl::Status InitCallElem(grpc_call_element* elem, |
|
|
|
|
@ -1864,32 +1874,25 @@ struct ChannelFilterMethods { |
|
|
|
|
static ArenaPromise<ServerMetadataHandle> MakeCallPromise( |
|
|
|
|
grpc_channel_element* elem, CallArgs call_args, |
|
|
|
|
NextPromiseFactory next_promise_factory) { |
|
|
|
|
return static_cast<ChannelFilter*>(elem->channel_data) |
|
|
|
|
->MakeCallPromise(std::move(call_args), |
|
|
|
|
std::move(next_promise_factory)); |
|
|
|
|
return ChannelFilterFromElem(elem)->MakeCallPromise( |
|
|
|
|
std::move(call_args), std::move(next_promise_factory)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void StartTransportOp(grpc_channel_element* elem, |
|
|
|
|
grpc_transport_op* op) { |
|
|
|
|
if (!static_cast<ChannelFilter*>(elem->channel_data) |
|
|
|
|
->StartTransportOp(op)) { |
|
|
|
|
if (!ChannelFilterFromElem(elem)->StartTransportOp(op)) { |
|
|
|
|
grpc_channel_next_op(elem, op); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void PostInitChannelElem(grpc_channel_stack*, |
|
|
|
|
grpc_channel_element* elem) { |
|
|
|
|
static_cast<ChannelFilter*>(elem->channel_data)->PostInit(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void DestroyChannelElem(grpc_channel_element* elem) { |
|
|
|
|
static_cast<ChannelFilter*>(elem->channel_data)->~ChannelFilter(); |
|
|
|
|
ChannelFilterFromElem(elem)->PostInit(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void GetChannelInfo(grpc_channel_element* elem, |
|
|
|
|
const grpc_channel_info* info) { |
|
|
|
|
if (!static_cast<ChannelFilter*>(elem->channel_data) |
|
|
|
|
->GetChannelInfo(info)) { |
|
|
|
|
if (!ChannelFilterFromElem(elem)->GetChannelInfo(info)) { |
|
|
|
|
grpc_channel_next_get_info(elem, info); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -1903,15 +1906,16 @@ struct ChannelFilterWithFlagsMethods { |
|
|
|
|
auto status = F::Create(args->channel_args, |
|
|
|
|
ChannelFilter::Args(args->channel_stack, elem)); |
|
|
|
|
if (!status.ok()) { |
|
|
|
|
static_assert( |
|
|
|
|
sizeof(promise_filter_detail::InvalidChannelFilter) <= sizeof(F), |
|
|
|
|
"InvalidChannelFilter must fit in F"); |
|
|
|
|
new (elem->channel_data) promise_filter_detail::InvalidChannelFilter(); |
|
|
|
|
new (elem->channel_data) F*(nullptr); |
|
|
|
|
return absl_status_to_grpc_error(status.status()); |
|
|
|
|
} |
|
|
|
|
new (elem->channel_data) F(std::move(*status)); |
|
|
|
|
new (elem->channel_data) F*(status->release()); |
|
|
|
|
return absl::OkStatus(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void DestroyChannelElem(grpc_channel_element* elem) { |
|
|
|
|
CreatedType<F> channel_elem(DownCast<F*>(ChannelFilterFromElem(elem))); |
|
|
|
|
} |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
} // namespace promise_filter_detail
|
|
|
|
|
@ -1957,7 +1961,8 @@ MakePromiseBasedFilter(const char* name) { |
|
|
|
|
// post_init_channel_elem
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::PostInitChannelElem, |
|
|
|
|
// destroy_channel_elem
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::DestroyChannelElem, |
|
|
|
|
promise_filter_detail::ChannelFilterWithFlagsMethods< |
|
|
|
|
F, kFlags>::DestroyChannelElem, |
|
|
|
|
// get_channel_info
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::GetChannelInfo, |
|
|
|
|
// name
|
|
|
|
|
@ -2003,7 +2008,8 @@ MakePromiseBasedFilter(const char* name) { |
|
|
|
|
// post_init_channel_elem
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::PostInitChannelElem, |
|
|
|
|
// destroy_channel_elem
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::DestroyChannelElem, |
|
|
|
|
promise_filter_detail::ChannelFilterWithFlagsMethods< |
|
|
|
|
F, kFlags>::DestroyChannelElem, |
|
|
|
|
// get_channel_info
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::GetChannelInfo, |
|
|
|
|
// name
|
|
|
|
|
@ -2045,7 +2051,8 @@ MakePromiseBasedFilter(const char* name) { |
|
|
|
|
// post_init_channel_elem
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::PostInitChannelElem, |
|
|
|
|
// destroy_channel_elem
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::DestroyChannelElem, |
|
|
|
|
promise_filter_detail::ChannelFilterWithFlagsMethods< |
|
|
|
|
F, kFlags>::DestroyChannelElem, |
|
|
|
|
// get_channel_info
|
|
|
|
|
promise_filter_detail::ChannelFilterMethods::GetChannelInfo, |
|
|
|
|
// name
|
|
|
|
|
|
Tanvi Jagtap
11 months ago
committed by
GitHub