| 
					
				 | 
			
			
				@@ -0,0 +1,373 @@ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "GRPCCall.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#include <grpc.h> 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#include <support/time.h> 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "GRPCMethodName.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/GRPCChannel.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/GRPCCompletionQueue.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/GRPCDelegateWrapper.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/GRPCMethodName+HTTP2Encoding.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/NSData+GRPC.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/NSDictionary+GRPC.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#import "private/NSError+GRPC.h" 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// A grpc_call_error represents a precondition failure when invoking the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// grpc_call_* functions. If one ever happens, it's a bug in this library. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// TODO(jcanizales): Can an application shut down gracefully when a thread other 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// than the main one throws an exception? 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+static void AssertNoErrorInCall(grpc_call_error error) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if (error != GRPC_CALL_OK) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    @throw [NSException exceptionWithName:NSInternalInconsistencyException 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                   reason:@"Precondition of grpc_call_* not met." 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                 userInfo:nil]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+@interface GRPCCall () <GRXWriteable> 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Makes it readwrite. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+@property(atomic, strong) NSDictionary *responseMetadata; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+@end 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// The following methods of a C gRPC call object aren't reentrant, and thus 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// calls to them must be serialized: 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// - add_metadata 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// - invoke 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// - start_write 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// - writes_done 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// - start_read 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// - destroy 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// The first four are called as part of responding to client commands, but 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// start_read we want to call as soon as we're notified that the RPC was 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// successfully established (which happens concurrently in the network queue). 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Serialization is achieved by using a private serial queue to operate the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// call object. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Because add_metadata and invoke are called and return successfully before 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// any of the other methods is called, they don't need to use the queue. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Furthermore, start_write and writes_done can only be called after the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// WRITE_ACCEPTED event for any previous write is received. This is achieved by 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// pausing the requests writer immediately every time it writes a value, and 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// resuming it again when WRITE_ACCEPTED is received. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Similarly, start_read can only be called after the READ event for any 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// previous read is received. This is easier to enforce, as we're writing the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// received messages into the writeable: start_read is enqueued once upon receiving 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// the CLIENT_METADATA_READ event, and then once after receiving each READ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// event. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+@implementation GRPCCall { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  dispatch_queue_t _callQueue; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_call *_gRPCCall; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  dispatch_once_t _callAlreadyInvoked; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  GRPCChannel *_channel; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  GRPCCompletionQueue *_completionQueue; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // The C gRPC library has less guarantees on the ordering of events than we 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // do. Particularly, in the face of errors, there's no ordering guarantee at 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // all. This wrapper over our actual writeable ensures thread-safety and 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // correct ordering. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  GRPCDelegateWrapper *_responseWriteable; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  id<GRXWriter> _requestWriter; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+@synthesize state = _state; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (instancetype)init { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  return [self initWithHost:nil method:nil requestsWriter:nil]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Designated initializer 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (instancetype)initWithHost:(NSString *)host 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                      method:(GRPCMethodName *)method 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+              requestsWriter:(id<GRXWriter>)requestWriter { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if (!host || !method) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [NSException raise:NSInvalidArgumentException format:@"Neither host nor method can be nil."]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // TODO(jcanizales): Throw if the requestWriter was already started. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if ((self = [super init])) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    static dispatch_once_t initialization; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    dispatch_once(&initialization, ^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      grpc_init(); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    }); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    _completionQueue = [GRPCCompletionQueue completionQueue]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    _channel = [GRPCChannel channelToHost:host]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    _gRPCCall = grpc_channel_create_call_old(_channel.unmanagedChannel, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                             method.HTTP2Path.UTF8String, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                             host.UTF8String, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                             gpr_inf_future); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // Serial queue to invoke the non-reentrant methods of the grpc_call object. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    _callQueue = dispatch_queue_create("org.grpc.call", NULL); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    _requestWriter = requestWriter; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  return self; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#pragma mark Finish 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)finishWithError:(NSError *)errorOrNil { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  _requestWriter.state = GRXWriterStateFinished; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  _requestWriter = nil; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if (errorOrNil) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [_responseWriteable cancelWithError:errorOrNil]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } else { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [_responseWriteable enqueueSuccessfulCompletion]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)cancelCall { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Can be called from any thread, any number of times. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  AssertNoErrorInCall(grpc_call_cancel(_gRPCCall)); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)cancel { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  [self finishWithError:[NSError errorWithDomain:kGRPCErrorDomain 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                            code:GRPCErrorCodeCancelled 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                        userInfo:nil]]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  [self cancelCall]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)dealloc { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_call *gRPCCall = _gRPCCall; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  dispatch_async(_callQueue, ^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    grpc_call_destroy(gRPCCall); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  }); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#pragma mark Read messages 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Only called from the call queue. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// The handler will be called from the network queue. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)startReadWithHandler:(GRPCEventHandler)handler { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  AssertNoErrorInCall(grpc_call_start_read_old(_gRPCCall, (__bridge_retained void *)handler)); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Called initially from the network queue once response headers are received, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// then "recursively" from the responseWriteable queue after each response from the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// server has been written. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// If the call is currently paused, this is a noop. Restarting the call will invoke this 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// method. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// TODO(jcanizales): Rename to readResponseIfNotPaused. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)startNextRead { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if (self.state == GRXWriterStatePaused) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  __weak GRPCCall *weakSelf = self; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  __weak GRPCDelegateWrapper *weakWriteable = _responseWriteable; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  dispatch_async(_callQueue, ^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [weakSelf startReadWithHandler:^(grpc_event *event) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      if (!event->data.read) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // No more responses from the server. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      NSData *data = [NSData grpc_dataWithByteBuffer:event->data.read]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      if (!data) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // The app doesn't have enough memory to hold the server response. We 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // don't want to throw, because the app shouldn't crash for a behavior 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // that's on the hands of any server to have. Instead we finish and ask 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // the server to cancel. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // TODO(jcanizales): No canonical code is appropriate for this situation 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // (because it's just a client problem). Use another domain and an 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        // appropriately-documented code. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        [weakSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                      code:GRPCErrorCodeInternal 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                  userInfo:nil]]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        [weakSelf cancelCall]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [weakWriteable enqueueMessage:data completionHandler:^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        [weakSelf startNextRead]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      }]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    }]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  }); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#pragma mark Send headers 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)addHeaderWithName:(NSString *)name binaryValue:(NSData *)value { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_metadata metadata; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Safe to discard const qualifiers; we're not going to modify the contents. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  metadata.key = (char *)name.UTF8String; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  metadata.value = (char *)value.bytes; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  metadata.value_length = value.length; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_call_add_metadata_old(_gRPCCall, &metadata, 0); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)addHeaderWithName:(NSString *)name ASCIIValue:(NSString *)value { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_metadata metadata; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Safe to discard const qualifiers; we're not going to modify the contents. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  metadata.key = (char *)name.UTF8String; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  metadata.value = (char *)value.UTF8String; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // The trailing \0 isn't encoded in HTTP2. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  metadata.value_length = value.length; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_call_add_metadata_old(_gRPCCall, &metadata, 0); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// TODO(jcanizales): Rename to commitHeaders. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)sendHeaders:(NSDictionary *)metadata { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  for (NSString *name in metadata) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    id value = metadata[name]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    if ([value isKindOfClass:[NSData class]]) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [self addHeaderWithName:name binaryValue:value]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    } else if ([value isKindOfClass:[NSString class]]) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [self addHeaderWithName:name ASCIIValue:value]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#pragma mark GRXWriteable implementation 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Only called from the call queue. The error handler will be called from the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// network queue if the write didn't succeed. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)writeMessage:(NSData *)message withErrorHandler:(void (^)())errorHandler { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  __weak GRPCCall *weakSelf = self; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  GRPCEventHandler resumingHandler = ^(grpc_event *event) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    if (event->data.write_accepted != GRPC_OP_OK) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      errorHandler(); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // Resume the request writer (even in the case of error). 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // TODO(jcanizales): No need to do it in the case of errors anymore? 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    GRPCCall *strongSelf = weakSelf; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    if (strongSelf) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      strongSelf->_requestWriter.state = GRXWriterStateStarted; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  }; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_byte_buffer *buffer = message.grpc_byteBuffer; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  AssertNoErrorInCall(grpc_call_start_write_old(_gRPCCall, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                buffer, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                (__bridge_retained void *)resumingHandler, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                0)); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  grpc_byte_buffer_destroy(buffer); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)didReceiveValue:(id)value { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // TODO(jcanizales): Throw/assert if value isn't NSData. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Pause the input and only resume it when the C layer notifies us that writes 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // can proceed. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  _requestWriter.state = GRXWriterStatePaused; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  __weak GRPCCall *weakSelf = self; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  dispatch_async(_callQueue, ^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [weakSelf writeMessage:value withErrorHandler:^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [weakSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                    code:GRPCErrorCodeInternal 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                userInfo:nil]]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    }]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  }); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Only called from the call queue. The error handler will be called from the 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// network queue if the requests stream couldn't be closed successfully. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)finishRequestWithErrorHandler:(void (^)())errorHandler { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  GRPCEventHandler handler = ^(grpc_event *event) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    if (event->data.finish_accepted != GRPC_OP_OK) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      errorHandler(); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  }; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  AssertNoErrorInCall(grpc_call_writes_done_old(_gRPCCall, (__bridge_retained void *)handler)); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)didFinishWithError:(NSError *)errorOrNil { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if (errorOrNil) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [self cancel]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } else { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    __weak GRPCCall *weakSelf = self; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    dispatch_async(_callQueue, ^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [weakSelf finishRequestWithErrorHandler:^{ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        [weakSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                      code:GRPCErrorCodeInternal 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                  userInfo:nil]]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      }]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    }); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#pragma mark Invoke 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// Both handlers will eventually be called, from the network queue. Writes can start immediately 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// after this. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// The first one (metadataHandler), when the response headers are received. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+// The second one (completionHandler), whenever the RPC finishes for any reason. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)invokeCallWithMetadataHandler:(GRPCEventHandler)metadataHandler 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                    completionHandler:(GRPCEventHandler)completionHandler { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  AssertNoErrorInCall(grpc_call_invoke_old(_gRPCCall, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                           _completionQueue.unmanagedQueue, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                           (__bridge_retained void *)metadataHandler, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                           (__bridge_retained void *)completionHandler, 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                           0)); 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)invokeCall { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  __weak GRPCCall *weakSelf = self; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  [self invokeCallWithMetadataHandler:^(grpc_event *event) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // Response metadata received. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // TODO(jcanizales): Name the type of event->data.client_metadata_read 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // in the C library so one can actually pass the object to a method. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    grpc_metadata *entries = event->data.client_metadata_read.elements; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    size_t count = event->data.client_metadata_read.count; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    GRPCCall *strongSelf = weakSelf; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    if (strongSelf) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      strongSelf.responseMetadata = [NSDictionary grpc_dictionaryFromMetadata:entries 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+                                                                        count:count]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [strongSelf startNextRead]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } completionHandler:^(grpc_event *event) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    // TODO(jcanizales): Merge HTTP2 trailers into response metadata. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    [weakSelf finishWithError:[NSError grpc_errorFromStatus:&event->data.finished]]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  }]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Now that the RPC has been initiated, request writes can start. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  [_requestWriter startWithWriteable:self]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+#pragma mark GRXWriter implementation 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)startWithWriteable:(id<GRXWriteable>)writeable { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // The following produces a retain cycle self:_responseWriteable:self, which is only 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // broken when didFinishWithError: is sent to the wrapped writeable. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Care is taken not to retain self strongly in any of the blocks used in 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // the implementation of GRPCCall, so that the life of the instance is 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // determined by this retain cycle. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  _responseWriteable = [[GRPCDelegateWrapper alloc] initWithWriteable:writeable writer:self]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  [self sendHeaders:_requestMetadata]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  [self invokeCall]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+- (void)setState:(GRXWriterState)newState { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  // Manual transitions are only allowed from the started or paused states. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  if (_state == GRXWriterStateNotStarted || _state == GRXWriterStateFinished) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+ 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  switch (newState) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    case GRXWriterStateFinished: 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      _state = newState; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      // Per GRXWriter's contract, setting the state to Finished manually 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      // means one doesn't wish the writeable to be messaged anymore. 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      [_responseWriteable cancelSilently]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      _responseWriteable = nil; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    case GRXWriterStatePaused: 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      _state = newState; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    case GRXWriterStateStarted: 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      if (_state == GRXWriterStatePaused) { 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        _state = newState; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+        [self startNextRead]; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+    case GRXWriterStateNotStarted: 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+      return; 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+  } 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+} 
			 | 
		
	
		
			
				 | 
				 | 
			
			
				+@end 
			 |