1 // Copyright 2024 gRPC authors.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 #include <grpc/support/port_platform.h>
16
17 #include "src/core/lib/transport/call_spine.h"
18
19 namespace grpc_core {
20
ForwardCall(CallHandler call_handler,CallInitiator call_initiator,ClientMetadataHandle client_initial_metadata)21 void ForwardCall(CallHandler call_handler, CallInitiator call_initiator,
22 ClientMetadataHandle client_initial_metadata) {
23 // Send initial metadata.
24 call_initiator.SpawnGuarded(
25 "send_initial_metadata",
26 [client_initial_metadata = std::move(client_initial_metadata),
27 call_initiator]() mutable {
28 return call_initiator.PushClientInitialMetadata(
29 std::move(client_initial_metadata));
30 });
31 // Read messages from handler into initiator.
32 call_handler.SpawnGuarded("read_messages", [call_handler,
33 call_initiator]() mutable {
34 return Seq(ForEach(OutgoingMessages(call_handler),
35 [call_initiator](MessageHandle msg) mutable {
36 // Need to spawn a job into the initiator's activity to
37 // push the message in.
38 return call_initiator.SpawnWaitable(
39 "send_message",
40 [msg = std::move(msg), call_initiator]() mutable {
41 return call_initiator.CancelIfFails(
42 call_initiator.PushMessage(std::move(msg)));
43 });
44 }),
45 [call_initiator](StatusFlag result) mutable {
46 if (result.ok()) {
47 call_initiator.SpawnInfallible(
48 "finish-downstream-ok", [call_initiator]() mutable {
49 call_initiator.FinishSends();
50 return Empty{};
51 });
52 } else {
53 call_initiator.SpawnInfallible("finish-downstream-fail",
54 [call_initiator]() mutable {
55 call_initiator.Cancel();
56 return Empty{};
57 });
58 }
59 return result;
60 });
61 });
62 call_initiator.SpawnInfallible("read_the_things", [call_initiator,
63 call_handler]() mutable {
64 return Seq(
65 call_initiator.CancelIfFails(TrySeq(
66 call_initiator.PullServerInitialMetadata(),
67 [call_handler,
68 call_initiator](absl::optional<ServerMetadataHandle> md) mutable {
69 const bool has_md = md.has_value();
70 call_handler.SpawnGuarded(
71 "recv_initial_metadata",
72 [md = std::move(md), call_handler]() mutable {
73 return call_handler.PushServerInitialMetadata(
74 std::move(md));
75 });
76 return If(
77 has_md,
78 ForEach(OutgoingMessages(call_initiator),
79 [call_handler](MessageHandle msg) mutable {
80 return call_handler.SpawnWaitable(
81 "recv_message",
82 [msg = std::move(msg), call_handler]() mutable {
83 return call_handler.CancelIfFails(
84 call_handler.PushMessage(std::move(msg)));
85 });
86 }),
87 []() -> StatusFlag { return Success{}; });
88 })),
89 call_initiator.PullServerTrailingMetadata(),
90 [call_handler](ServerMetadataHandle md) mutable {
91 call_handler.SpawnGuarded(
92 "recv_trailing_metadata",
93 [md = std::move(md), call_handler]() mutable {
94 return call_handler.PushServerTrailingMetadata(std::move(md));
95 });
96 return Empty{};
97 });
98 });
99 }
100
MakeCall(grpc_event_engine::experimental::EventEngine * event_engine,Arena * arena)101 CallInitiatorAndHandler MakeCall(
102 grpc_event_engine::experimental::EventEngine* event_engine, Arena* arena) {
103 auto spine = CallSpine::Create(event_engine, arena);
104 return {CallInitiator(spine), CallHandler(spine)};
105 }
106
107 } // namespace grpc_core
108