Merge pull request #16190 from muxi/config-isolation

Implement L38: gRPC Objective-C API Upgrade

3 files changed, 59 insertions, 56 deletions

diff --git a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.cc b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.cc
index 40a30e4a31..6e08d27b21 100644
--- a/src/core/ext/transport/cronet/client/secure/cronet_channel_create.cc
+++ b/src/core/ext/transport/cronet/client/secure/cronet_channel_create.cc
@@ -46,9 +46,21 @@ GRPCAPI grpc_channel* grpc_cronet_secure_channel_create(
           "grpc_create_cronet_transport: stream_engine = %p, target=%s", engine,
           target);
 
+  // Disable client authority filter when using Cronet
+  grpc_arg disable_client_authority_filter_arg;
+  disable_client_authority_filter_arg.key =
+      const_cast<char*>(GRPC_ARG_DISABLE_CLIENT_AUTHORITY_FILTER);
+  disable_client_authority_filter_arg.type = GRPC_ARG_INTEGER;
+  disable_client_authority_filter_arg.value.integer = 1;
+  grpc_channel_args* new_args = grpc_channel_args_copy_and_add(
+      args, &disable_client_authority_filter_arg, 1);
+
   grpc_transport* ct =
-      grpc_create_cronet_transport(engine, target, args, reserved);
+      grpc_create_cronet_transport(engine, target, new_args, reserved);
 
   grpc_core::ExecCtx exec_ctx;
-  return grpc_channel_create(target, args, GRPC_CLIENT_DIRECT_CHANNEL, ct);
+  grpc_channel* channel =
+      grpc_channel_create(target, new_args, GRPC_CLIENT_DIRECT_CHANNEL, ct);
+  grpc_channel_args_destroy(new_args);
+  return channel;
 }
diff --git a/src/core/lib/iomgr/cfstream_handle.cc b/src/core/lib/iomgr/cfstream_handle.cc
index 827fd24831..6cb9ca1a0d 100644
--- a/src/core/lib/iomgr/cfstream_handle.cc
+++ b/src/core/lib/iomgr/cfstream_handle.cc
@@ -52,62 +52,52 @@ CFStreamHandle* CFStreamHandle::CreateStreamHandle(
 void CFStreamHandle::ReadCallback(CFReadStreamRef stream,
                                   CFStreamEventType type,
                                   void* client_callback_info) {
+  grpc_core::ExecCtx exec_ctx;
   CFStreamHandle* handle = static_cast<CFStreamHandle*>(client_callback_info);
-  CFSTREAM_HANDLE_REF(handle, "read callback");
-  dispatch_async(
-      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
-        grpc_core::ExecCtx exec_ctx;
-        if (grpc_tcp_trace.enabled()) {
-          gpr_log(GPR_DEBUG, "CFStream ReadCallback (%p, %p, %lu, %p)", handle,
-                  stream, type, client_callback_info);
-        }
-        switch (type) {
-          case kCFStreamEventOpenCompleted:
-            handle->open_event_.SetReady();
-            break;
-          case kCFStreamEventHasBytesAvailable:
-          case kCFStreamEventEndEncountered:
-            handle->read_event_.SetReady();
-            break;
-          case kCFStreamEventErrorOccurred:
-            handle->open_event_.SetReady();
-            handle->read_event_.SetReady();
-            break;
-          default:
-            GPR_UNREACHABLE_CODE(return );
-        }
-        CFSTREAM_HANDLE_UNREF(handle, "read callback");
-      });
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "CFStream ReadCallback (%p, %p, %lu, %p)", handle,
+            stream, type, client_callback_info);
+  }
+  switch (type) {
+    case kCFStreamEventOpenCompleted:
+      handle->open_event_.SetReady();
+      break;
+    case kCFStreamEventHasBytesAvailable:
+    case kCFStreamEventEndEncountered:
+      handle->read_event_.SetReady();
+      break;
+    case kCFStreamEventErrorOccurred:
+      handle->open_event_.SetReady();
+      handle->read_event_.SetReady();
+      break;
+    default:
+      GPR_UNREACHABLE_CODE(return );
+  }
 }
 void CFStreamHandle::WriteCallback(CFWriteStreamRef stream,
                                    CFStreamEventType type,
                                    void* clientCallBackInfo) {
+  grpc_core::ExecCtx exec_ctx;
   CFStreamHandle* handle = static_cast<CFStreamHandle*>(clientCallBackInfo);
-  CFSTREAM_HANDLE_REF(handle, "write callback");
-  dispatch_async(
-      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
-        grpc_core::ExecCtx exec_ctx;
-        if (grpc_tcp_trace.enabled()) {
-          gpr_log(GPR_DEBUG, "CFStream WriteCallback (%p, %p, %lu, %p)", handle,
-                  stream, type, clientCallBackInfo);
-        }
-        switch (type) {
-          case kCFStreamEventOpenCompleted:
-            handle->open_event_.SetReady();
-            break;
-          case kCFStreamEventCanAcceptBytes:
-          case kCFStreamEventEndEncountered:
-            handle->write_event_.SetReady();
-            break;
-          case kCFStreamEventErrorOccurred:
-            handle->open_event_.SetReady();
-            handle->write_event_.SetReady();
-            break;
-          default:
-            GPR_UNREACHABLE_CODE(return );
-        }
-        CFSTREAM_HANDLE_UNREF(handle, "write callback");
-      });
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "CFStream WriteCallback (%p, %p, %lu, %p)", handle,
+            stream, type, clientCallBackInfo);
+  }
+  switch (type) {
+    case kCFStreamEventOpenCompleted:
+      handle->open_event_.SetReady();
+      break;
+    case kCFStreamEventCanAcceptBytes:
+    case kCFStreamEventEndEncountered:
+      handle->write_event_.SetReady();
+      break;
+    case kCFStreamEventErrorOccurred:
+      handle->open_event_.SetReady();
+      handle->write_event_.SetReady();
+      break;
+    default:
+      GPR_UNREACHABLE_CODE(return );
+  }
 }
 
 CFStreamHandle::CFStreamHandle(CFReadStreamRef read_stream,
@@ -116,6 +106,7 @@ CFStreamHandle::CFStreamHandle(CFReadStreamRef read_stream,
   open_event_.InitEvent();
   read_event_.InitEvent();
   write_event_.InitEvent();
+  dispatch_queue_ = dispatch_queue_create(nullptr, DISPATCH_QUEUE_SERIAL);
   CFStreamClientContext ctx = {0, static_cast<void*>(this),
                                CFStreamHandle::Retain, CFStreamHandle::Release,
                                nil};
@@ -129,10 +120,8 @@ CFStreamHandle::CFStreamHandle(CFReadStreamRef read_stream,
       kCFStreamEventOpenCompleted | kCFStreamEventCanAcceptBytes |
           kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered,
       CFStreamHandle::WriteCallback, &ctx);
-  CFReadStreamScheduleWithRunLoop(read_stream, CFRunLoopGetMain(),
-                                  kCFRunLoopCommonModes);
-  CFWriteStreamScheduleWithRunLoop(write_stream, CFRunLoopGetMain(),
-                                   kCFRunLoopCommonModes);
+  CFReadStreamSetDispatchQueue(read_stream, dispatch_queue_);
+  CFWriteStreamSetDispatchQueue(write_stream, dispatch_queue_);
 }
 
 CFStreamHandle::~CFStreamHandle() {
diff --git a/src/core/lib/iomgr/cfstream_handle.h b/src/core/lib/iomgr/cfstream_handle.h
index 4258e72431..93ec5f044b 100644
--- a/src/core/lib/iomgr/cfstream_handle.h
+++ b/src/core/lib/iomgr/cfstream_handle.h
@@ -62,6 +62,8 @@ class CFStreamHandle final {
   grpc_core::LockfreeEvent read_event_;
   grpc_core::LockfreeEvent write_event_;
 
+  dispatch_queue_t dispatch_queue_;
+
   gpr_refcount refcount_;
 };