xref: /aosp_15_r20/external/grpc-grpc/src/core/lib/transport/call_spine.cc (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2)
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