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 "src/core/client_channel/direct_channel.h"
16
17 #include "src/core/config/core_configuration.h"
18 #include "src/core/lib/event_engine/event_engine_context.h"
19 #include "src/core/lib/surface/channel_stack_type.h"
20 #include "src/core/lib/surface/client_call.h"
21 #include "src/core/lib/transport/interception_chain.h"
22 #include "src/core/util/orphanable.h"
23
24 namespace grpc_core {
25
Create(std::string target,const ChannelArgs & args)26 absl::StatusOr<RefCountedPtr<DirectChannel>> DirectChannel::Create(
27 std::string target, const ChannelArgs& args) {
28 auto* transport = args.GetObject<Transport>();
29 if (transport == nullptr) {
30 return absl::InvalidArgumentError("Transport not set in ChannelArgs");
31 }
32 if (transport->client_transport() == nullptr) {
33 return absl::InvalidArgumentError("Transport is not a client transport");
34 }
35 auto transport_call_destination = MakeRefCounted<TransportCallDestination>(
36 OrphanablePtr<ClientTransport>(transport->client_transport()));
37 auto event_engine =
38 args.GetObjectRef<grpc_event_engine::experimental::EventEngine>();
39 if (event_engine == nullptr) {
40 return absl::InvalidArgumentError("EventEngine not set in ChannelArgs");
41 }
42 InterceptionChainBuilder builder(args);
43 CoreConfiguration::Get().channel_init().AddToInterceptionChainBuilder(
44 GRPC_CLIENT_DIRECT_CHANNEL, builder);
45 auto interception_chain = builder.Build(transport_call_destination);
46 if (!interception_chain.ok()) return interception_chain.status();
47 return MakeRefCounted<DirectChannel>(
48 std::move(target), args, std::move(event_engine),
49 std::move(transport_call_destination), std::move(*interception_chain));
50 }
51
Orphaned()52 void DirectChannel::Orphaned() {
53 transport_call_destination_.reset();
54 interception_chain_.reset();
55 }
56
StartCall(UnstartedCallHandler unstarted_handler)57 void DirectChannel::StartCall(UnstartedCallHandler unstarted_handler) {
58 unstarted_handler.SpawnInfallible(
59 "start",
60 [interception_chain = interception_chain_, unstarted_handler]() mutable {
61 interception_chain->StartCall(std::move(unstarted_handler));
62 return []() {};
63 });
64 }
65
GetInfo(const grpc_channel_info *)66 void DirectChannel::GetInfo(const grpc_channel_info*) {
67 // TODO(roth): Implement this.
68 }
69
CreateCall(grpc_call * parent_call,uint32_t propagation_mask,grpc_completion_queue * cq,grpc_pollset_set *,Slice path,absl::optional<Slice> authority,Timestamp deadline,bool)70 grpc_call* DirectChannel::CreateCall(
71 grpc_call* parent_call, uint32_t propagation_mask,
72 grpc_completion_queue* cq, grpc_pollset_set* /*pollset_set_alternative*/,
73 Slice path, absl::optional<Slice> authority, Timestamp deadline,
74 bool /*registered_method*/) {
75 auto arena = call_arena_allocator()->MakeArena();
76 arena->SetContext<grpc_event_engine::experimental::EventEngine>(
77 event_engine_.get());
78 return MakeClientCall(parent_call, propagation_mask, cq, std::move(path),
79 std::move(authority), false, deadline,
80 compression_options(), std::move(arena), Ref());
81 }
82
83 } // namespace grpc_core
84