1 // Copyright 2022 The 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 #include "src/core/lib/iomgr/event_engine_shims/tcp_client.h"
15
16 #include <grpc/event_engine/event_engine.h>
17 #include <grpc/support/port_platform.h>
18 #include <grpc/support/time.h>
19
20 #include "absl/status/status.h"
21 #include "absl/status/statusor.h"
22 #include "src/core/lib/address_utils/sockaddr_utils.h"
23 #include "src/core/lib/debug/trace.h"
24 #include "src/core/lib/event_engine/default_event_engine.h"
25 #include "src/core/lib/event_engine/resolved_address_internal.h"
26 #include "src/core/lib/iomgr/closure.h"
27 #include "src/core/lib/iomgr/endpoint.h"
28 #include "src/core/lib/iomgr/error.h"
29 #include "src/core/lib/iomgr/event_engine_shims/endpoint.h"
30 #include "src/core/lib/iomgr/exec_ctx.h"
31 #include "src/core/lib/resource_quota/resource_quota.h"
32 #include "src/core/lib/transport/error_utils.h"
33
34 namespace grpc_event_engine {
35 namespace experimental {
36
event_engine_tcp_client_connect(grpc_closure * on_connect,grpc_endpoint ** endpoint,const grpc_event_engine::experimental::EndpointConfig & config,const grpc_resolved_address * addr,grpc_core::Timestamp deadline)37 int64_t event_engine_tcp_client_connect(
38 grpc_closure* on_connect, grpc_endpoint** endpoint,
39 const grpc_event_engine::experimental::EndpointConfig& config,
40 const grpc_resolved_address* addr, grpc_core::Timestamp deadline) {
41 auto resource_quota = reinterpret_cast<grpc_core::ResourceQuota*>(
42 config.GetVoidPointer(GRPC_ARG_RESOURCE_QUOTA));
43 auto addr_uri = grpc_sockaddr_to_uri(addr);
44 EventEngine* engine_ptr = reinterpret_cast<EventEngine*>(
45 config.GetVoidPointer(GRPC_INTERNAL_ARG_EVENT_ENGINE));
46 // Keeps the engine alive for some tests that have not otherwise instantiated
47 // an EventEngine
48 std::shared_ptr<EventEngine> keeper;
49 if (engine_ptr == nullptr) {
50 keeper = GetDefaultEventEngine();
51 engine_ptr = keeper.get();
52 }
53 EventEngine::ConnectionHandle handle = engine_ptr->Connect(
54 [on_connect,
55 endpoint](absl::StatusOr<std::unique_ptr<EventEngine::Endpoint>> ep) {
56 grpc_core::ApplicationCallbackExecCtx app_ctx;
57 grpc_core::ExecCtx exec_ctx;
58 absl::Status conn_status = ep.ok() ? absl::OkStatus() : ep.status();
59 if (ep.ok()) {
60 *endpoint = grpc_event_engine_endpoint_create(std::move(*ep));
61 } else {
62 *endpoint = nullptr;
63 }
64 GRPC_TRACE_LOG(event_engine, INFO)
65 << "EventEngine::Connect Status: " << ep.status();
66 grpc_core::ExecCtx::Run(DEBUG_LOCATION, on_connect,
67 absl_status_to_grpc_error(conn_status));
68 },
69 CreateResolvedAddress(*addr), config,
70 resource_quota != nullptr
71 ? resource_quota->memory_quota()->CreateMemoryOwner()
72 : grpc_event_engine::experimental::MemoryAllocator(),
73 std::max(grpc_core::Duration::Milliseconds(1),
74 deadline - grpc_core::Timestamp::Now()));
75 GRPC_TRACE_LOG(event_engine, INFO)
76 << "EventEngine::Connect Peer: " << *addr_uri << ", handle: " << handle;
77 return handle.keys[0];
78 }
79
event_engine_tcp_client_cancel_connect(int64_t connection_handle)80 bool event_engine_tcp_client_cancel_connect(int64_t connection_handle) {
81 GRPC_TRACE_LOG(event_engine, INFO)
82 << "EventEngine::CancelConnect handle: " << connection_handle;
83 return GetDefaultEventEngine()->CancelConnect(
84 {static_cast<intptr_t>(connection_handle), 0});
85 }
86 } // namespace experimental
87 } // namespace grpc_event_engine
88