src/core/lib/event_engine/cf_engine/cfstream_endpoint.cc - third_party/grpc - Git at Google

 // Copyright 2023 The gRPC Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 #include <grpc/support/port_platform.h>

 #ifdef GPR_APPLE

 #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"

 namespace grpc_event_engine {
 namespace experimental {

 namespace {

 int kDefaultReadBufferSize = 8192;

 absl::Status CFErrorToStatus(CFTypeUniqueRef<CFErrorRef> cf_error) {
   if (cf_error == nullptr) {
     return absl::OkStatus();
   }
   CFErrorDomain cf_domain = CFErrorGetDomain((cf_error));
   CFIndex code = CFErrorGetCode((cf_error));
   CFTypeUniqueRef<CFStringRef> cf_desc = CFErrorCopyDescription((cf_error));
   char domain_buf[256];
   char desc_buf[256];
   CFStringGetCString(cf_domain, domain_buf, 256, kCFStringEncodingUTF8);
   CFStringGetCString(cf_desc, desc_buf, 256, kCFStringEncodingUTF8);
   return absl::Status(absl::StatusCode::kUnknown,
                       absl::StrFormat("(domain:%s, code:%ld, description:%s)",
                                       domain_buf, code, desc_buf));
 }

 absl::StatusOr<EventEngine::ResolvedAddress> CFReadStreamLocallAddress(
     CFReadStreamRef stream) {
   CFTypeUniqueRef<CFDataRef> cf_native_handle = static_cast<CFDataRef>(
       CFReadStreamCopyProperty(stream, kCFStreamPropertySocketNativeHandle));
   CFSocketNativeHandle socket;
   CFDataGetBytes(cf_native_handle, CFRangeMake(0, sizeof(CFSocketNativeHandle)),
                  (UInt8*)&socket);
   EventEngine::ResolvedAddress addr;
   socklen_t len = EventEngine::ResolvedAddress::MAX_SIZE_BYTES;
   if (getsockname(socket, const_cast<sockaddr*>(addr.address()), &len) < 0) {
     return absl::InternalError(
         absl::StrCat("getsockname:", grpc_core::StrError(errno)));
   }
   return EventEngine::ResolvedAddress(addr.address(), len);
 }

 }  // namespace

 bool CFStreamEndpointImpl::CancelConnect(absl::Status status) {
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
       "CFStreamEndpointImpl::CancelConnect: status: %s, this: %p",
       status.ToString().c_str(), this);

   return open_event_.SetShutdown(std::move(status));
 }

 void CFStreamEndpointImpl::Connect(
     absl::AnyInvocable<void(absl::Status)> on_connect,
     EventEngine::ResolvedAddress addr) {
   auto addr_uri = ResolvedAddressToURI(addr);

   if (!addr_uri.ok()) {
     on_connect(std::move(addr_uri).status());
     return;
   }

   GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Connect: %s",
                                    addr_uri.value().c_str());

   peer_address_ = std::move(addr);
   auto host_port = ResolvedAddressToNormalizedString(peer_address_);
   if (!host_port.ok()) {
     on_connect(std::move(host_port).status());
     return;
   }

   peer_address_string_ = host_port.value();
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
       "CFStreamEndpointImpl::Connect, host_port: %s", host_port->c_str());

   std::string host_string;
   std::string port_string;
   grpc_core::SplitHostPort(host_port.value(), &host_string, &port_string);
   CFStringRef host = CFStringCreateWithCString(NULL, host_string.c_str(),
                                                kCFStringEncodingUTF8);
   int port = ResolvedAddressGetPort(peer_address_);
   CFStreamCreatePairWithSocketToHost(NULL, host, port, &cf_read_stream_,
                                      &cf_write_stream_);

   CFStreamClientContext cf_context = {0, this, Retain, Release, nullptr};
   CFReadStreamSetClient(
       cf_read_stream_,
       kCFStreamEventOpenCompleted | kCFStreamEventHasBytesAvailable |
           kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered,
       ReadCallback, &cf_context);
   CFWriteStreamSetClient(
       cf_write_stream_,
       kCFStreamEventOpenCompleted | kCFStreamEventCanAcceptBytes |
           kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered,
       WriteCallback, &cf_context);
   CFReadStreamSetDispatchQueue(cf_read_stream_,
                                dispatch_get_global_queue(QOS_CLASS_DEFAULT, 0));
   CFWriteStreamSetDispatchQueue(
       cf_write_stream_, dispatch_get_global_queue(QOS_CLASS_DEFAULT, 0));

   if (!CFReadStreamOpen(cf_read_stream_)) {
     auto status = CFErrorToStatus(CFReadStreamCopyError(cf_read_stream_));
     on_connect(std::move(status));
     return;
   }

   if (!CFWriteStreamOpen(cf_write_stream_)) {
     auto status = CFErrorToStatus(CFWriteStreamCopyError(cf_write_stream_));
     on_connect(std::move(status));
     return;
   }

   open_event_.NotifyOn(new PosixEngineClosure(
       [that = Ref(),
        on_connect = std::move(on_connect)](absl::Status status) mutable {
         if (!status.ok()) {
           on_connect(std::move(status));
           return;
         }

         auto local_addr = CFReadStreamLocallAddress(that->cf_read_stream_);
         if (!local_addr.ok()) {
           on_connect(std::move(local_addr).status());
           return;
         }

         that->local_address_ = local_addr.value();
         that->local_address_string_ =
             *ResolvedAddressToURI(that->local_address_);
         on_connect(absl::OkStatus());
       },
       false /* is_permanent */));
 }

 /* static */ void CFStreamEndpointImpl::ReadCallback(
     CFReadStreamRef stream, CFStreamEventType type,
     void* client_callback_info) {
   auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info);

   GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
       "CFStreamEndpointImpl::ReadCallback, type: %lu, this: %p", type, self);

   switch (type) {
     case kCFStreamEventOpenCompleted:
       // wait for write stream open completed to signal connection ready
       break;
     case kCFStreamEventHasBytesAvailable:
       ABSL_FALLTHROUGH_INTENDED;
     case kCFStreamEventEndEncountered:
       self->read_event_.SetReady();
       break;
     case kCFStreamEventErrorOccurred: {
       auto status = CFErrorToStatus(CFReadStreamCopyError(stream));
       GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream Read error: %s",
                                        status.ToString().c_str());

       self->open_event_.SetShutdown(status);
       self->read_event_.SetShutdown(status);
       self->write_event_.SetShutdown(status);
     } break;
     default:
       GPR_UNREACHABLE_CODE(return);
   }
 }

 /* static */
 void CFStreamEndpointImpl::WriteCallback(CFWriteStreamRef stream,
                                          CFStreamEventType type,
                                          void* client_callback_info) {
   auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info);
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
       "CFStreamEndpointImpl::WriteCallback, type: %lu, this: %p", type, self);

   switch (type) {
     case kCFStreamEventOpenCompleted:
       self->open_event_.SetReady();
       break;
     case kCFStreamEventCanAcceptBytes:
       ABSL_FALLTHROUGH_INTENDED;
     case kCFStreamEventEndEncountered:
       self->write_event_.SetReady();
       break;
     case kCFStreamEventErrorOccurred: {
       auto status = CFErrorToStatus(CFWriteStreamCopyError(stream));
       GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream Write error: %s",
                                        status.ToString().c_str());

       self->open_event_.SetShutdown(status);
       self->read_event_.SetShutdown(status);
       self->write_event_.SetShutdown(status);
     } break;
     default:
       GPR_UNREACHABLE_CODE(return);
   }
 }

 CFStreamEndpointImpl::CFStreamEndpointImpl(
     std::shared_ptr<CFEventEngine> engine, MemoryAllocator memory_allocator)
     : engine_(std::move(engine)),
       memory_allocator_(std::move(memory_allocator)),
       open_event_(engine_.get()),
       read_event_(engine_.get()),
       write_event_(engine_.get()) {
   open_event_.InitEvent();
   read_event_.InitEvent();
   write_event_.InitEvent();
 }

 CFStreamEndpointImpl::~CFStreamEndpointImpl() {
   open_event_.DestroyEvent();
   read_event_.DestroyEvent();
   write_event_.DestroyEvent();
 }

 void CFStreamEndpointImpl::Shutdown() {
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Shutdown: this: %p",
                                    this);

   auto shutdownStatus =
       absl::Status(absl::StatusCode::kUnknown,
                    absl::StrFormat("Shutting down CFStreamEndpointImpl"));
   open_event_.SetShutdown(shutdownStatus);
   read_event_.SetShutdown(shutdownStatus);
   write_event_.SetShutdown(shutdownStatus);

   CFReadStreamSetClient(cf_read_stream_, kCFStreamEventNone, nullptr, nullptr);
   CFWriteStreamSetClient(cf_write_stream_, kCFStreamEventNone, nullptr,
                          nullptr);
   CFReadStreamClose(cf_read_stream_);
   CFWriteStreamClose(cf_write_stream_);
 }

 bool CFStreamEndpointImpl::Read(
     absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer,
     const EventEngine::Endpoint::ReadArgs* /* args */) {
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Read, this: %p",
                                    this);

   read_event_.NotifyOn(new PosixEngineClosure(
       [that = Ref(), on_read = std::move(on_read),
        buffer](absl::Status status) mutable {
         if (status.ok()) {
           that->DoRead(std::move(on_read), buffer);
         } else {
           on_read(status);
         }
       },
       false /* is_permanent*/));

   return false;
 }

 void CFStreamEndpointImpl::DoRead(
     absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer) {
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::DoRead, this: %p",
                                    this);

   auto buffer_index = buffer->AppendIndexed(
       Slice(memory_allocator_.MakeSlice(kDefaultReadBufferSize)));

   CFIndex read_size = CFReadStreamRead(
       cf_read_stream_,
       internal::SliceCast<MutableSlice>(buffer->MutableSliceAt(buffer_index))
           .begin(),
       kDefaultReadBufferSize);

   if (read_size < 0) {
     auto status = CFErrorToStatus(CFReadStreamCopyError(cf_read_stream_));
     GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream read error: %s, read_size: %ld",
                                      status.ToString().c_str(), read_size);
     on_read(status);
     return;
   }

   buffer->RemoveLastNBytes(buffer->Length() - read_size);
   on_read(absl::OkStatus());
 }

 bool CFStreamEndpointImpl::Write(
     absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data,
     const EventEngine::Endpoint::WriteArgs* /* args */) {
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Write, this: %p",
                                    this);

   write_event_.NotifyOn(new PosixEngineClosure(
       [that = Ref(), on_writable = std::move(on_writable),
        data](absl::Status status) mutable {
         if (status.ok()) {
           that->DoWrite(std::move(on_writable), data);
         } else {
           on_writable(status);
         }
       },
       false /* is_permanent*/));

   return false;
 }

 void CFStreamEndpointImpl::DoWrite(
     absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data) {
   GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::DoWrite, this: %p",
                                    this);

   size_t total_written_size = 0;
   for (size_t i = 0; i < data->Count(); i++) {
     auto slice = data->RefSlice(i);
     size_t written_size =
         CFWriteStreamWrite(cf_write_stream_, slice.begin(), slice.size());

     total_written_size += written_size;
     if (written_size < slice.size()) {
       SliceBuffer written;
       data->MoveFirstNBytesIntoSliceBuffer(total_written_size, written);

       write_event_.NotifyOn(new PosixEngineClosure(
           [that = Ref(), on_writable = std::move(on_writable),
            data](absl::Status status) mutable {
             if (status.ok()) {
               that->DoWrite(std::move(on_writable), data);
             } else {
               on_writable(status);
             }
           },
           false /* is_permanent*/));
       return;
     }
   }
   on_writable(absl::OkStatus());
 }

 }  // namespace experimental
 }  // namespace grpc_event_engine

 #endif  // GPR_APPLE
	// Copyright 2023 The gRPC Authors
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	#include <grpc/support/port_platform.h>

	#ifdef GPR_APPLE

	#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"

	namespace grpc_event_engine {
	namespace experimental {

	namespace {

	int kDefaultReadBufferSize = 8192;

	absl::Status CFErrorToStatus(CFTypeUniqueRef<CFErrorRef> cf_error) {
	if (cf_error == nullptr) {
	return absl::OkStatus();
	}
	CFErrorDomain cf_domain = CFErrorGetDomain((cf_error));
	CFIndex code = CFErrorGetCode((cf_error));
	CFTypeUniqueRef<CFStringRef> cf_desc = CFErrorCopyDescription((cf_error));
	char domain_buf[256];
	char desc_buf[256];
	CFStringGetCString(cf_domain, domain_buf, 256, kCFStringEncodingUTF8);
	CFStringGetCString(cf_desc, desc_buf, 256, kCFStringEncodingUTF8);
	return absl::Status(absl::StatusCode::kUnknown,
	absl::StrFormat("(domain:%s, code:%ld, description:%s)",
	domain_buf, code, desc_buf));
	}

	absl::StatusOr<EventEngine::ResolvedAddress> CFReadStreamLocallAddress(
	CFReadStreamRef stream) {
	CFTypeUniqueRef<CFDataRef> cf_native_handle = static_cast<CFDataRef>(
	CFReadStreamCopyProperty(stream, kCFStreamPropertySocketNativeHandle));
	CFSocketNativeHandle socket;
	CFDataGetBytes(cf_native_handle, CFRangeMake(0, sizeof(CFSocketNativeHandle)),
	(UInt8*)&socket);
	EventEngine::ResolvedAddress addr;
	socklen_t len = EventEngine::ResolvedAddress::MAX_SIZE_BYTES;
	if (getsockname(socket, const_cast<sockaddr*>(addr.address()), &len) < 0) {
	return absl::InternalError(
	absl::StrCat("getsockname:", grpc_core::StrError(errno)));
	}
	return EventEngine::ResolvedAddress(addr.address(), len);
	}

	} // namespace

	bool CFStreamEndpointImpl::CancelConnect(absl::Status status) {
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
	"CFStreamEndpointImpl::CancelConnect: status: %s, this: %p",
	status.ToString().c_str(), this);

	return open_event_.SetShutdown(std::move(status));
	}

	void CFStreamEndpointImpl::Connect(
	absl::AnyInvocable<void(absl::Status)> on_connect,
	EventEngine::ResolvedAddress addr) {
	auto addr_uri = ResolvedAddressToURI(addr);

	if (!addr_uri.ok()) {
	on_connect(std::move(addr_uri).status());
	return;
	}

	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Connect: %s",
	addr_uri.value().c_str());

	peer_address_ = std::move(addr);
	auto host_port = ResolvedAddressToNormalizedString(peer_address_);
	if (!host_port.ok()) {
	on_connect(std::move(host_port).status());
	return;
	}

	peer_address_string_ = host_port.value();
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
	"CFStreamEndpointImpl::Connect, host_port: %s", host_port->c_str());

	std::string host_string;
	std::string port_string;
	grpc_core::SplitHostPort(host_port.value(), &host_string, &port_string);
	CFStringRef host = CFStringCreateWithCString(NULL, host_string.c_str(),
	kCFStringEncodingUTF8);
	int port = ResolvedAddressGetPort(peer_address_);
	CFStreamCreatePairWithSocketToHost(NULL, host, port, &cf_read_stream_,
	&cf_write_stream_);

	CFStreamClientContext cf_context = {0, this, Retain, Release, nullptr};
	CFReadStreamSetClient(
	cf_read_stream_,
	kCFStreamEventOpenCompleted \| kCFStreamEventHasBytesAvailable \|
	kCFStreamEventErrorOccurred \| kCFStreamEventEndEncountered,
	ReadCallback, &cf_context);
	CFWriteStreamSetClient(
	cf_write_stream_,
	kCFStreamEventOpenCompleted \| kCFStreamEventCanAcceptBytes \|
	kCFStreamEventErrorOccurred \| kCFStreamEventEndEncountered,
	WriteCallback, &cf_context);
	CFReadStreamSetDispatchQueue(cf_read_stream_,
	dispatch_get_global_queue(QOS_CLASS_DEFAULT, 0));
	CFWriteStreamSetDispatchQueue(
	cf_write_stream_, dispatch_get_global_queue(QOS_CLASS_DEFAULT, 0));

	if (!CFReadStreamOpen(cf_read_stream_)) {
	auto status = CFErrorToStatus(CFReadStreamCopyError(cf_read_stream_));
	on_connect(std::move(status));
	return;
	}

	if (!CFWriteStreamOpen(cf_write_stream_)) {
	auto status = CFErrorToStatus(CFWriteStreamCopyError(cf_write_stream_));
	on_connect(std::move(status));
	return;
	}

	open_event_.NotifyOn(new PosixEngineClosure(
	[that = Ref(),
	on_connect = std::move(on_connect)](absl::Status status) mutable {
	if (!status.ok()) {
	on_connect(std::move(status));
	return;
	}

	auto local_addr = CFReadStreamLocallAddress(that->cf_read_stream_);
	if (!local_addr.ok()) {
	on_connect(std::move(local_addr).status());
	return;
	}

	that->local_address_ = local_addr.value();
	that->local_address_string_ =
	*ResolvedAddressToURI(that->local_address_);
	on_connect(absl::OkStatus());
	},
	false /* is_permanent */));
	}

	/* static */ void CFStreamEndpointImpl::ReadCallback(
	CFReadStreamRef stream, CFStreamEventType type,
	void* client_callback_info) {
	auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info);

	GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
	"CFStreamEndpointImpl::ReadCallback, type: %lu, this: %p", type, self);

	switch (type) {
	case kCFStreamEventOpenCompleted:
	// wait for write stream open completed to signal connection ready
	break;
	case kCFStreamEventHasBytesAvailable:
	ABSL_FALLTHROUGH_INTENDED;
	case kCFStreamEventEndEncountered:
	self->read_event_.SetReady();
	break;
	case kCFStreamEventErrorOccurred: {
	auto status = CFErrorToStatus(CFReadStreamCopyError(stream));
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream Read error: %s",
	status.ToString().c_str());

	self->open_event_.SetShutdown(status);
	self->read_event_.SetShutdown(status);
	self->write_event_.SetShutdown(status);
	} break;
	default:
	GPR_UNREACHABLE_CODE(return);
	}
	}

	/* static */
	void CFStreamEndpointImpl::WriteCallback(CFWriteStreamRef stream,
	CFStreamEventType type,
	void* client_callback_info) {
	auto self = static_cast<CFStreamEndpointImpl*>(client_callback_info);
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE(
	"CFStreamEndpointImpl::WriteCallback, type: %lu, this: %p", type, self);

	switch (type) {
	case kCFStreamEventOpenCompleted:
	self->open_event_.SetReady();
	break;
	case kCFStreamEventCanAcceptBytes:
	ABSL_FALLTHROUGH_INTENDED;
	case kCFStreamEventEndEncountered:
	self->write_event_.SetReady();
	break;
	case kCFStreamEventErrorOccurred: {
	auto status = CFErrorToStatus(CFWriteStreamCopyError(stream));
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream Write error: %s",
	status.ToString().c_str());

	self->open_event_.SetShutdown(status);
	self->read_event_.SetShutdown(status);
	self->write_event_.SetShutdown(status);
	} break;
	default:
	GPR_UNREACHABLE_CODE(return);
	}
	}

	CFStreamEndpointImpl::CFStreamEndpointImpl(
	std::shared_ptr<CFEventEngine> engine, MemoryAllocator memory_allocator)
	: engine_(std::move(engine)),
	memory_allocator_(std::move(memory_allocator)),
	open_event_(engine_.get()),
	read_event_(engine_.get()),
	write_event_(engine_.get()) {
	open_event_.InitEvent();
	read_event_.InitEvent();
	write_event_.InitEvent();
	}

	CFStreamEndpointImpl::~CFStreamEndpointImpl() {
	open_event_.DestroyEvent();
	read_event_.DestroyEvent();
	write_event_.DestroyEvent();
	}

	void CFStreamEndpointImpl::Shutdown() {
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Shutdown: this: %p",
	this);

	auto shutdownStatus =
	absl::Status(absl::StatusCode::kUnknown,
	absl::StrFormat("Shutting down CFStreamEndpointImpl"));
	open_event_.SetShutdown(shutdownStatus);
	read_event_.SetShutdown(shutdownStatus);
	write_event_.SetShutdown(shutdownStatus);

	CFReadStreamSetClient(cf_read_stream_, kCFStreamEventNone, nullptr, nullptr);
	CFWriteStreamSetClient(cf_write_stream_, kCFStreamEventNone, nullptr,
	nullptr);
	CFReadStreamClose(cf_read_stream_);
	CFWriteStreamClose(cf_write_stream_);
	}

	bool CFStreamEndpointImpl::Read(
	absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer,
	const EventEngine::Endpoint::ReadArgs* /* args */) {
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Read, this: %p",
	this);

	read_event_.NotifyOn(new PosixEngineClosure(
	[that = Ref(), on_read = std::move(on_read),
	buffer](absl::Status status) mutable {
	if (status.ok()) {
	that->DoRead(std::move(on_read), buffer);
	} else {
	on_read(status);
	}
	},
	false /* is_permanent*/));

	return false;
	}

	void CFStreamEndpointImpl::DoRead(
	absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer) {
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::DoRead, this: %p",
	this);

	auto buffer_index = buffer->AppendIndexed(
	Slice(memory_allocator_.MakeSlice(kDefaultReadBufferSize)));

	CFIndex read_size = CFReadStreamRead(
	cf_read_stream_,
	internal::SliceCast<MutableSlice>(buffer->MutableSliceAt(buffer_index))
	.begin(),
	kDefaultReadBufferSize);

	if (read_size < 0) {
	auto status = CFErrorToStatus(CFReadStreamCopyError(cf_read_stream_));
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStream read error: %s, read_size: %ld",
	status.ToString().c_str(), read_size);
	on_read(status);
	return;
	}

	buffer->RemoveLastNBytes(buffer->Length() - read_size);
	on_read(absl::OkStatus());
	}

	bool CFStreamEndpointImpl::Write(
	absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data,
	const EventEngine::Endpoint::WriteArgs* /* args */) {
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::Write, this: %p",
	this);

	write_event_.NotifyOn(new PosixEngineClosure(
	[that = Ref(), on_writable = std::move(on_writable),
	data](absl::Status status) mutable {
	if (status.ok()) {
	that->DoWrite(std::move(on_writable), data);
	} else {
	on_writable(status);
	}
	},
	false /* is_permanent*/));

	return false;
	}

	void CFStreamEndpointImpl::DoWrite(
	absl::AnyInvocable<void(absl::Status)> on_writable, SliceBuffer* data) {
	GRPC_EVENT_ENGINE_ENDPOINT_TRACE("CFStreamEndpointImpl::DoWrite, this: %p",
	this);

	size_t total_written_size = 0;
	for (size_t i = 0; i < data->Count(); i++) {
	auto slice = data->RefSlice(i);
	size_t written_size =
	CFWriteStreamWrite(cf_write_stream_, slice.begin(), slice.size());

	total_written_size += written_size;
	if (written_size < slice.size()) {
	SliceBuffer written;
	data->MoveFirstNBytesIntoSliceBuffer(total_written_size, written);

	write_event_.NotifyOn(new PosixEngineClosure(
	[that = Ref(), on_writable = std::move(on_writable),
	data](absl::Status status) mutable {
	if (status.ok()) {
	that->DoWrite(std::move(on_writable), data);
	} else {
	on_writable(status);
	}
	},
	false /* is_permanent*/));
	return;
	}
	}
	on_writable(absl::OkStatus());
	}

	} // namespace experimental
	} // namespace grpc_event_engine

	#endif // GPR_APPLE