|
|
|
@ -23,7 +23,6 @@ |
|
|
|
#include "absl/strings/str_format.h" |
|
|
|
#include "absl/strings/str_format.h" |
|
|
|
|
|
|
|
|
|
|
|
#include "src/core/lib/event_engine/cf_engine/cfstream_endpoint.h" |
|
|
|
#include "src/core/lib/event_engine/cf_engine/cfstream_endpoint.h" |
|
|
|
#include "src/core/lib/event_engine/trace.h" |
|
|
|
|
|
|
|
#include "src/core/lib/gprpp/strerror.h" |
|
|
|
#include "src/core/lib/gprpp/strerror.h" |
|
|
|
|
|
|
|
|
|
|
|
namespace grpc_event_engine { |
|
|
|
namespace grpc_event_engine { |
|
|
|
@ -68,9 +67,9 @@ absl::StatusOr<EventEngine::ResolvedAddress> CFReadStreamLocallAddress( |
|
|
|
} // namespace
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
|
|
|
|
bool CFStreamEndpointImpl::CancelConnect(absl::Status status) { |
|
|
|
bool CFStreamEndpointImpl::CancelConnect(absl::Status status) { |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE( |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
"CFStreamEndpointImpl::CancelConnect: status: %s, this: %p", |
|
|
|
<< "CFStreamEndpointImpl::CancelConnect: status: " << status |
|
|
|
status.ToString().c_str(), this); |
|
|
|
<< ", this: " << this; |
|
|
|
|
|
|
|
|
|
|
|
return open_event_.SetShutdown(std::move(status)); |
|
|
|
return open_event_.SetShutdown(std::move(status)); |
|
|
|
} |
|
|
|
} |
|
|
|
@ -85,8 +84,8 @@ void CFStreamEndpointImpl::Connect( |
|
|
|
return; |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Connect: %s", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
addr_uri.value().c_str()); |
|
|
|
<< "CFStreamEndpointImpl::Connect: " << addr_uri.value(); |
|
|
|
|
|
|
|
|
|
|
|
peer_address_ = std::move(addr); |
|
|
|
peer_address_ = std::move(addr); |
|
|
|
auto host_port = ResolvedAddressToNormalizedString(peer_address_); |
|
|
|
auto host_port = ResolvedAddressToNormalizedString(peer_address_); |
|
|
|
@ -96,8 +95,8 @@ void CFStreamEndpointImpl::Connect( |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
peer_address_string_ = host_port.value(); |
|
|
|
peer_address_string_ = host_port.value(); |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE( |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
"CFStreamEndpointImpl::Connect, host_port: %s", host_port->c_str()); |
|
|
|
<< "CFStreamEndpointImpl::Connect, host_port: " << peer_address_string_; |
|
|
|
|
|
|
|
|
|
|
|
std::string host_string; |
|
|
|
std::string host_string; |
|
|
|
std::string port_string; |
|
|
|
std::string port_string; |
|
|
|
@ -163,8 +162,9 @@ void CFStreamEndpointImpl::Connect( |
|
|
|
void* client_callback_info) { |
|
|
|
void* client_callback_info) { |
|
|
|
auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info); |
|
|
|
auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info); |
|
|
|
|
|
|
|
|
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE( |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
"CFStreamEndpointImpl::ReadCallback, type: %lu, this: %p", type, self); |
|
|
|
<< "CFStreamEndpointImpl::ReadCallback, type: " << type |
|
|
|
|
|
|
|
<< ", this: " << self; |
|
|
|
|
|
|
|
|
|
|
|
switch (type) { |
|
|
|
switch (type) { |
|
|
|
case kCFStreamEventOpenCompleted: |
|
|
|
case kCFStreamEventOpenCompleted: |
|
|
|
@ -177,8 +177,8 @@ void CFStreamEndpointImpl::Connect( |
|
|
|
break; |
|
|
|
break; |
|
|
|
case kCFStreamEventErrorOccurred: { |
|
|
|
case kCFStreamEventErrorOccurred: { |
|
|
|
auto status = CFErrorToStatus(CFReadStreamCopyError(stream)); |
|
|
|
auto status = CFErrorToStatus(CFReadStreamCopyError(stream)); |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream Read error: %s", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
status.ToString().c_str()); |
|
|
|
<< "CFStream Read error: " << status; |
|
|
|
|
|
|
|
|
|
|
|
self->open_event_.SetShutdown(status); |
|
|
|
self->open_event_.SetShutdown(status); |
|
|
|
self->read_event_.SetShutdown(status); |
|
|
|
self->read_event_.SetShutdown(status); |
|
|
|
@ -194,8 +194,9 @@ void CFStreamEndpointImpl::WriteCallback(CFWriteStreamRef stream, |
|
|
|
CFStreamEventType type, |
|
|
|
CFStreamEventType type, |
|
|
|
void* client_callback_info) { |
|
|
|
void* client_callback_info) { |
|
|
|
auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info); |
|
|
|
auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info); |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE( |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
"CFStreamEndpointImpl::WriteCallback, type: %lu, this: %p", type, self); |
|
|
|
<< "CFStreamEndpointImpl::WriteCallback, type: " << type |
|
|
|
|
|
|
|
<< ", this: " << self; |
|
|
|
|
|
|
|
|
|
|
|
switch (type) { |
|
|
|
switch (type) { |
|
|
|
case kCFStreamEventOpenCompleted: |
|
|
|
case kCFStreamEventOpenCompleted: |
|
|
|
@ -208,8 +209,8 @@ void CFStreamEndpointImpl::WriteCallback(CFWriteStreamRef stream, |
|
|
|
break; |
|
|
|
break; |
|
|
|
case kCFStreamEventErrorOccurred: { |
|
|
|
case kCFStreamEventErrorOccurred: { |
|
|
|
auto status = CFErrorToStatus(CFWriteStreamCopyError(stream)); |
|
|
|
auto status = CFErrorToStatus(CFWriteStreamCopyError(stream)); |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream Write error: %s", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
status.ToString().c_str()); |
|
|
|
<< "CFStream Write error: " << status; |
|
|
|
|
|
|
|
|
|
|
|
self->open_event_.SetShutdown(status); |
|
|
|
self->open_event_.SetShutdown(status); |
|
|
|
self->read_event_.SetShutdown(status); |
|
|
|
self->read_event_.SetShutdown(status); |
|
|
|
@ -239,8 +240,8 @@ CFStreamEndpointImpl::~CFStreamEndpointImpl() { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
void CFStreamEndpointImpl::Shutdown() { |
|
|
|
void CFStreamEndpointImpl::Shutdown() { |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Shutdown: this: %p", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
this); |
|
|
|
<< "CFStreamEndpointImpl::Shutdown: this: " << this; |
|
|
|
|
|
|
|
|
|
|
|
auto shutdownStatus = |
|
|
|
auto shutdownStatus = |
|
|
|
absl::Status(absl::StatusCode::kUnknown, |
|
|
|
absl::Status(absl::StatusCode::kUnknown, |
|
|
|
@ -259,8 +260,8 @@ void CFStreamEndpointImpl::Shutdown() { |
|
|
|
bool CFStreamEndpointImpl::Read( |
|
|
|
bool CFStreamEndpointImpl::Read( |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer, |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer, |
|
|
|
const EventEngine::Endpoint::ReadArgs* /* args */) { |
|
|
|
const EventEngine::Endpoint::ReadArgs* /* args */) { |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Read, this: %p", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
this); |
|
|
|
<< "CFStreamEndpointImpl::Read, this: " << this; |
|
|
|
|
|
|
|
|
|
|
|
read_event_.NotifyOn(new PosixEngineClosure( |
|
|
|
read_event_.NotifyOn(new PosixEngineClosure( |
|
|
|
[that = Ref(), on_read = std::move(on_read), |
|
|
|
[that = Ref(), on_read = std::move(on_read), |
|
|
|
@ -278,8 +279,8 @@ bool CFStreamEndpointImpl::Read( |
|
|
|
|
|
|
|
|
|
|
|
void CFStreamEndpointImpl::DoRead( |
|
|
|
void CFStreamEndpointImpl::DoRead( |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer) { |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer) { |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::DoRead, this: %p", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
this); |
|
|
|
<< "CFStreamEndpointImpl::DoRead, this: " << this; |
|
|
|
|
|
|
|
|
|
|
|
auto buffer_index = buffer->AppendIndexed( |
|
|
|
auto buffer_index = buffer->AppendIndexed( |
|
|
|
Slice(memory_allocator_.MakeSlice(kDefaultReadBufferSize))); |
|
|
|
Slice(memory_allocator_.MakeSlice(kDefaultReadBufferSize))); |
|
|
|
@ -292,8 +293,8 @@ void CFStreamEndpointImpl::DoRead( |
|
|
|
|
|
|
|
|
|
|
|
if (read_size < 0) { |
|
|
|
if (read_size < 0) { |
|
|
|
auto status = CFErrorToStatus(CFReadStreamCopyError(cf_read_stream_)); |
|
|
|
auto status = CFErrorToStatus(CFReadStreamCopyError(cf_read_stream_)); |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream read error: %s, read_size: %ld", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
status.ToString().c_str(), read_size); |
|
|
|
<< "CFStream read error: " << status << ", read_size: " << read_size; |
|
|
|
on_read(status); |
|
|
|
on_read(status); |
|
|
|
return; |
|
|
|
return; |
|
|
|
} |
|
|
|
} |
|
|
|
@ -305,8 +306,8 @@ void CFStreamEndpointImpl::DoRead( |
|
|
|
bool CFStreamEndpointImpl::Write( |
|
|
|
bool CFStreamEndpointImpl::Write( |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data, |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data, |
|
|
|
const EventEngine::Endpoint::WriteArgs* /* args */) { |
|
|
|
const EventEngine::Endpoint::WriteArgs* /* args */) { |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Write, this: %p", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
this); |
|
|
|
<< "CFStreamEndpointImpl::Write, this: " << this; |
|
|
|
|
|
|
|
|
|
|
|
write_event_.NotifyOn(new PosixEngineClosure( |
|
|
|
write_event_.NotifyOn(new PosixEngineClosure( |
|
|
|
[that = Ref(), on_writable = std::move(on_writable), |
|
|
|
[that = Ref(), on_writable = std::move(on_writable), |
|
|
|
@ -324,8 +325,8 @@ bool CFStreamEndpointImpl::Write( |
|
|
|
|
|
|
|
|
|
|
|
void CFStreamEndpointImpl::DoWrite( |
|
|
|
void CFStreamEndpointImpl::DoWrite( |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data) { |
|
|
|
absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data) { |
|
|
|
GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::DoWrite, this: %p", |
|
|
|
GRPC_TRACE_LOG(event_engine_endpoint, INFO) |
|
|
|
this); |
|
|
|
<< "CFStreamEndpointImpl::DoWrite, this: " << this; |
|
|
|
|
|
|
|
|
|
|
|
size_t total_written_size = 0; |
|
|
|
size_t total_written_size = 0; |
|
|
|
for (size_t i = 0; i < data->Count(); i++) { |
|
|
|
for (size_t i = 0; i < data->Count(); i++) { |
|
|
|
|
AJ Heller
9 months ago
committed by
Copybara-Service