Imports code of the generic gRPC client library.
diff --git a/src/objective-c/GRPCClient/GRPCCall.h b/src/objective-c/GRPCClient/GRPCCall.h
new file mode 100644
index 0000000..db138fd
--- /dev/null
+++ b/src/objective-c/GRPCClient/GRPCCall.h
@@ -0,0 +1,57 @@
+#import <Foundation/Foundation.h>
+#import <RxLibrary/GRXWriter.h>
+
+@class GRPCMethodName;
+
+@class GRPCCall;
+
+// The gRPC protocol is an RPC protocol on top of HTTP2.
+//
+// While the most common type of RPC receives only one request message and
+// returns only one response message, the protocol also supports RPCs that
+// return multiple individual messages in a streaming fashion, RPCs that
+// accept a stream of request messages, or RPCs with both streaming requests
+// and responses.
+//
+// Conceptually, each gRPC call consists of a bidirectional stream of binary
+// messages, with RPCs of the "non-streaming type" sending only one message in
+// the corresponding direction (the protocol doesn't make any distinction).
+//
+// Each RPC uses a different HTTP2 stream, and thus multiple simultaneous RPCs
+// can be multiplexed transparently on the same TCP connection.
+@interface GRPCCall : NSObject<GRXWriter>
+
+// These HTTP2 headers will be passed to the server as part of this call. Each
+// HTTP2 header is a name-value pair with string names and either string or binary values.
+// The passed dictionary has to use NSString keys, corresponding to the header names. The
+// value associated to each can be a NSString object or a NSData object. E.g.:
+//
+// call.requestMetadata = @{
+// @"Authorization": @"Bearer ...",
+// @"SomeBinaryHeader": someData
+// };
+//
+// After the call is started, modifying this won't have any effect.
+@property(nonatomic, readwrite) NSMutableDictionary *requestMetadata;
+
+// This isn't populated until the first event is delivered to the handler.
+@property(atomic, readonly) NSDictionary *responseMetadata;
+
+// The request writer has to write NSData objects into the provided Writeable. The server will
+// receive each of those separately and in order.
+// A gRPC call might not complete until the request writer finishes. On the other hand, the
+// request finishing doesn't necessarily make the call to finish, as the server might continue
+// sending messages to the response side of the call indefinitely (depending on the semantics of
+// the specific remote method called).
+// To finish a call right away, invoke cancel.
+- (instancetype)initWithHost:(NSString *)host
+ method:(GRPCMethodName *)method
+ requestsWriter:(id<GRXWriter>)requestsWriter NS_DESIGNATED_INITIALIZER;
+
+// Finishes the request side of this call, notifies the server that the RPC
+// should be cancelled, and finishes the response side of the call with an error
+// of code CANCELED.
+- (void)cancel;
+
+// TODO(jcanizales): Let specify a deadline. As a category of GRXWriter?
+@end
diff --git a/src/objective-c/GRPCClient/GRPCCall.m b/src/objective-c/GRPCClient/GRPCCall.m
new file mode 100644
index 0000000..b9248be
--- /dev/null
+++ b/src/objective-c/GRPCClient/GRPCCall.m
@@ -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
diff --git a/src/objective-c/GRPCClient/GRPCMethodName.h b/src/objective-c/GRPCClient/GRPCMethodName.h
new file mode 100644
index 0000000..4fb86d2
--- /dev/null
+++ b/src/objective-c/GRPCClient/GRPCMethodName.h
@@ -0,0 +1,15 @@
+#import <Foundation/Foundation.h>
+
+// See the README file for an introduction to this library.
+
+// A fully-qualified gRPC method name. Full qualification is needed because a gRPC endpoint can
+// implement multiple interfaces.
+// TODO(jcanizales): Is this proto-specific, or actual part of gRPC? If the former, move one layer up.
+@interface GRPCMethodName : NSObject
+@property(nonatomic, readonly) NSString *package;
+@property(nonatomic, readonly) NSString *interface;
+@property(nonatomic, readonly) NSString *method;
+- (instancetype)initWithPackage:(NSString *)package
+ interface:(NSString *)interface
+ method:(NSString *)method;
+@end
diff --git a/src/objective-c/GRPCClient/GRPCMethodName.m b/src/objective-c/GRPCClient/GRPCMethodName.m
new file mode 100644
index 0000000..be9fd4b
--- /dev/null
+++ b/src/objective-c/GRPCClient/GRPCMethodName.m
@@ -0,0 +1,14 @@
+#import "GRPCMethodName.h"
+
+@implementation GRPCMethodName
+- (instancetype)initWithPackage:(NSString *)package
+ interface:(NSString *)interface
+ method:(NSString *)method {
+ if ((self = [super init])) {
+ _package = [package copy];
+ _interface = [interface copy];
+ _method = [method copy];
+ }
+ return self;
+}
+@end
diff --git a/src/objective-c/GRPCClient/README.md b/src/objective-c/GRPCClient/README.md
new file mode 100644
index 0000000..9b87f03
--- /dev/null
+++ b/src/objective-c/GRPCClient/README.md
@@ -0,0 +1,4 @@
+This is a generic gRPC client for Objective-C on iOS.
+
+If you're trying to get started with the library or with gRPC, you should first
+read GRPCCall.h.
diff --git a/src/objective-c/GRPCClient/private/GRPCChannel.h b/src/objective-c/GRPCClient/private/GRPCChannel.h
new file mode 100644
index 0000000..8772acc
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCChannel.h
@@ -0,0 +1,17 @@
+#import <Foundation/Foundation.h>
+
+struct grpc_channel;
+
+// Each separate instance of this class represents at least one TCP
+// connection to the provided host. To create a grpc_call, pass the
+// value of the unmanagedChannel property to grpc_channel_create_call.
+// Release this object when the call is finished.
+@interface GRPCChannel : NSObject
+@property(nonatomic, readonly) struct grpc_channel *unmanagedChannel;
+
+// Convenience constructor to allow for reuse of connections.
++ (instancetype)channelToHost:(NSString *)host;
+
+// Designated initializer
+- (instancetype)initWithHost:(NSString *)host;
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCChannel.m b/src/objective-c/GRPCClient/private/GRPCChannel.m
new file mode 100644
index 0000000..af4a01e
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCChannel.m
@@ -0,0 +1,32 @@
+#import "GRPCChannel.h"
+
+#import <grpc.h>
+
+@implementation GRPCChannel
+
++ (instancetype)channelToHost:(NSString *)host {
+ // TODO(jcanizales): Reuse channels.
+ return [[self alloc] initWithHost:host];
+}
+
+- (instancetype)init {
+ return [self initWithHost:nil];
+}
+
+// Designated initializer
+- (instancetype)initWithHost:(NSString *)host {
+ if (!host) {
+ [NSException raise:NSInvalidArgumentException format:@"Host can't be nil."];
+ }
+ if ((self = [super init])) {
+ _unmanagedChannel = grpc_channel_create(host.UTF8String, NULL);
+ }
+ return self;
+}
+
+- (void)dealloc {
+ // TODO(jcanizales): Be sure to add a test with a server that closes the connection prematurely,
+ // as in the past that made this call to crash.
+ grpc_channel_destroy(_unmanagedChannel);
+}
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
new file mode 100644
index 0000000..503df94
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
@@ -0,0 +1,21 @@
+#import <Foundation/Foundation.h>
+
+struct grpc_completion_queue;
+struct grpc_event;
+
+typedef void(^GRPCEventHandler)(struct grpc_event *event);
+
+// This class lets one more easily use grpc_completion_queue. To use it, pass
+// the value of the unmanagedQueue property of an instance of this class to
+// grpc_call_start_invoke. Then for every grpc_call_* method that accepts a tag,
+// you can pass a block of type GRPCEventHandler (remembering to cast it using
+// __bridge_retained). The block is guaranteed to eventually be called, by a
+// concurrent queue, and then released. Each such block is passed a pointer to
+// the grpc_event that carried it (in event->tag).
+// Release the GRPCCompletionQueue object only after you are not going to pass
+// any more blocks to the grpc_call that's using it.
+@interface GRPCCompletionQueue : NSObject
+@property(nonatomic, readonly) struct grpc_completion_queue *unmanagedQueue;
+
++ (instancetype)completionQueue;
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
new file mode 100644
index 0000000..d2508da
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
@@ -0,0 +1,73 @@
+#import "GRPCCompletionQueue.h"
+
+#import <grpc.h>
+
+@implementation GRPCCompletionQueue
+
++ (instancetype)completionQueue {
+ // TODO(jcanizales): Reuse completion queues to consume only one thread,
+ // instead of one per call.
+ return [[self alloc] init];
+}
+
+- (instancetype)init {
+ if ((self = [super init])) {
+ _unmanagedQueue = grpc_completion_queue_create();
+
+ // This is for the following block to capture the pointer by value (instead
+ // of retaining self and doing self->_unmanagedQueue). This is essential
+ // because the block doesn't end until after grpc_completion_queue_shutdown
+ // is called, and we only want that to happen after nobody's using the queue
+ // anymore (i.e. on self dealloc). So the block would never end if it
+ // retained self.
+ grpc_completion_queue *unmanagedQueue = _unmanagedQueue;
+
+ // Start a loop on a concurrent queue to read events from the completion
+ // queue and dispatch each.
+ static dispatch_once_t initialization;
+ static dispatch_queue_t gDefaultConcurrentQueue;
+ dispatch_once(&initialization, ^{
+ gDefaultConcurrentQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
+ });
+ dispatch_async(gDefaultConcurrentQueue, ^{
+ while (YES) {
+ // The following call blocks until an event is available.
+ grpc_event *event = grpc_completion_queue_next(unmanagedQueue, gpr_inf_future);
+ switch (event->type) {
+ case GRPC_WRITE_ACCEPTED:
+ case GRPC_FINISH_ACCEPTED:
+ case GRPC_CLIENT_METADATA_READ:
+ case GRPC_READ:
+ case GRPC_FINISHED:
+ if (event->tag) {
+ GRPCEventHandler handler = (__bridge_transfer GRPCEventHandler) event->tag;
+ handler(event);
+ }
+ grpc_event_finish(event);
+ continue;
+ case GRPC_QUEUE_SHUTDOWN:
+ grpc_completion_queue_destroy(unmanagedQueue);
+ grpc_event_finish(event);
+ return;
+ case GRPC_SERVER_RPC_NEW:
+ NSAssert(NO, @"C gRPC library produced a server-only event.");
+ continue;
+ }
+ // This means the C gRPC library produced an event that wasn't known
+ // when this library was written. To preserve evolvability, ignore the
+ // unknown event on release builds.
+ NSAssert(NO, @"C gRPC library produced an unknown event.");
+ };
+ });
+ }
+ return self;
+}
+
+- (void)dealloc {
+ // This makes the completion queue produce a GRPC_QUEUE_SHUTDOWN event *after*
+ // all other pending events are flushed. What this means is all the blocks
+ // passed to the gRPC C library as void* are eventually called, even if some
+ // are called after self is dealloc'd.
+ grpc_completion_queue_shutdown(_unmanagedQueue);
+}
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.h b/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.h
new file mode 100644
index 0000000..70a07f8
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.h
@@ -0,0 +1,48 @@
+#import <Foundation/Foundation.h>
+
+@protocol GRXWriteable;
+@protocol GRXWriter;
+
+// This is a thread-safe wrapper over a GRXWriteable instance. It lets one
+// enqueue calls to a GRXWriteable instance for the main thread, guaranteeing
+// that didFinishWithError: is the last message sent to it (no matter what
+// messages are sent to the wrapper, in what order, nor from which thread). It
+// also guarantees that, if cancelWithError: is called from the main thread
+// (e.g. by the app cancelling the writes), no further messages are sent to the
+// writeable except didFinishWithError:.
+//
+// TODO(jcanizales): Let the user specify another queue for the writeable
+// callbacks.
+// TODO(jcanizales): Rename to GRXWriteableWrapper and move to the Rx library.
+@interface GRPCDelegateWrapper : NSObject
+
+// The GRXWriteable passed is the wrapped writeable.
+// Both the GRXWriter instance and the GRXWriteable instance are retained until
+// didFinishWithError: is sent to the writeable, and released after that.
+// This is used to create a retain cycle that keeps both objects alive until the
+// writing is explicitly finished.
+- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable writer:(id<GRXWriter>)writer
+ NS_DESIGNATED_INITIALIZER;
+
+// Enqueues didReceiveValue: to be sent to the writeable in the main thread.
+// The passed handler is invoked from the main thread after didReceiveValue:
+// returns.
+- (void)enqueueMessage:(NSData *)message completionHandler:(void (^)())handler;
+
+// Enqueues didFinishWithError:nil to be sent to the writeable in the main
+// thread. After that message is sent to the writeable, all other methods of
+// this object are effectively noops.
+- (void)enqueueSuccessfulCompletion;
+
+// If the writeable has not yet received a didFinishWithError: message, this
+// will enqueue one to be sent to it in the main thread, and cancel all other
+// pending messages to the writeable enqueued by this object (both past and
+// future).
+// The error argument cannot be nil.
+- (void)cancelWithError:(NSError *)error;
+
+// Cancels all pending messages to the writeable enqueued by this object (both
+// past and future). Because the writeable won't receive didFinishWithError:,
+// this also releases the writeable and the writer.
+- (void)cancelSilently;
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.m b/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.m
new file mode 100644
index 0000000..7c64850
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCDelegateWrapper.m
@@ -0,0 +1,87 @@
+#import "GRPCDelegateWrapper.h"
+
+#import <net/grpc/objc/RxLibrary/GRXWriteable.h>
+
+@interface GRPCDelegateWrapper ()
+// These are atomic so that cancellation can nillify them from any thread.
+@property(atomic, strong) id<GRXWriteable> writeable;
+@property(atomic, strong) id<GRXWriter> writer;
+@end
+
+@implementation GRPCDelegateWrapper {
+ dispatch_queue_t _writeableQueue;
+ // This ensures that didFinishWithError: is only sent once to the writeable.
+ dispatch_once_t _alreadyFinished;
+}
+
+- (instancetype)init {
+ return [self initWithWriteable:nil writer:nil];
+}
+
+// Designated initializer
+- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable writer:(id<GRXWriter>)writer {
+ if (self = [super init]) {
+ _writeableQueue = dispatch_get_main_queue();
+ _writeable = writeable;
+ _writer = writer;
+ }
+ return self;
+}
+
+- (void)enqueueMessage:(NSData *)message completionHandler:(void (^)())handler {
+ dispatch_async(_writeableQueue, ^{
+ // We're racing a possible cancellation performed by another thread. To turn
+ // all already-enqueued messages into noops, cancellation nillifies the
+ // writeable property. If we get it before it's nil, we won
+ // the race.
+ id<GRXWriteable> writeable = self.writeable;
+ if (writeable) {
+ [writeable didReceiveValue:message];
+ handler();
+ }
+ });
+}
+
+- (void)enqueueSuccessfulCompletion {
+ dispatch_async(_writeableQueue, ^{
+ dispatch_once(&_alreadyFinished, ^{
+ // Cancellation is now impossible. None of the other three blocks can run
+ // concurrently with this one.
+ [self.writeable didFinishWithError:nil];
+ // Break the retain cycle with writer, and skip any possible message to the
+ // wrapped writeable enqueued after this one.
+ self.writeable = nil;
+ self.writer = nil;
+ });
+ });
+}
+
+- (void)cancelWithError:(NSError *)error {
+ NSAssert(error, @"For a successful completion, use enqueueSuccessfulCompletion.");
+ dispatch_once(&_alreadyFinished, ^{
+ // Skip any of the still-enqueued messages to the wrapped writeable. We use
+ // the atomic setter to nillify writer and writeable because we might be
+ // running concurrently with the blocks in _writeableQueue, and assignment
+ // with ARC isn't atomic.
+ id<GRXWriteable> writeable = self.writeable;
+ self.writeable = nil;
+
+ dispatch_async(_writeableQueue, ^{
+ [writeable didFinishWithError:error];
+ // Break the retain cycle with writer.
+ self.writer = nil;
+ });
+ });
+}
+
+- (void)cancelSilently {
+ dispatch_once(&_alreadyFinished, ^{
+ // Skip any of the still-enqueued messages to the wrapped writeable. We use
+ // the atomic setter to nillify writer and writeable because we might be
+ // running concurrently with the blocks in _writeableQueue, and assignment
+ // with ARC isn't atomic.
+ self.writeable = nil;
+ self.writer = nil;
+ });
+}
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCMethodName+HTTP2Encoding.h b/src/objective-c/GRPCClient/private/GRPCMethodName+HTTP2Encoding.h
new file mode 100644
index 0000000..504c669
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCMethodName+HTTP2Encoding.h
@@ -0,0 +1,7 @@
+#import <Foundation/Foundation.h>
+
+#import "GRPCMethodName.h"
+
+@interface GRPCMethodName (HTTP2Encoding)
+- (NSString *)HTTP2Path;
+@end
diff --git a/src/objective-c/GRPCClient/private/GRPCMethodName+HTTP2Encoding.m b/src/objective-c/GRPCClient/private/GRPCMethodName+HTTP2Encoding.m
new file mode 100644
index 0000000..2e9fe8d
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/GRPCMethodName+HTTP2Encoding.m
@@ -0,0 +1,11 @@
+#import "GRPCMethodName+HTTP2Encoding.h"
+
+@implementation GRPCMethodName (HTTP2Encoding)
+- (NSString *)HTTP2Path {
+ if (self.package) {
+ return [NSString stringWithFormat:@"/%@.%@/%@", self.package, self.interface, self.method];
+ } else {
+ return [NSString stringWithFormat:@"/%@/%@", self.interface, self.method];
+ }
+}
+@end
diff --git a/src/objective-c/GRPCClient/private/NSData+GRPC.h b/src/objective-c/GRPCClient/private/NSData+GRPC.h
new file mode 100644
index 0000000..8cb7b76
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/NSData+GRPC.h
@@ -0,0 +1,8 @@
+#import <Foundation/Foundation.h>
+
+struct grpc_byte_buffer;
+
+@interface NSData (GRPC)
++ (instancetype)grpc_dataWithByteBuffer:(struct grpc_byte_buffer *)buffer;
+- (struct grpc_byte_buffer *)grpc_byteBuffer;
+@end
diff --git a/src/objective-c/GRPCClient/private/NSData+GRPC.m b/src/objective-c/GRPCClient/private/NSData+GRPC.m
new file mode 100644
index 0000000..47f7a07
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/NSData+GRPC.m
@@ -0,0 +1,53 @@
+#import "NSData+GRPC.h"
+
+#include <byte_buffer.h>
+#include <string.h>
+
+// TODO(jcanizales): Move these two incantations to the C library.
+
+static void CopyByteBufferToCharArray(grpc_byte_buffer *buffer, char *array) {
+ size_t offset = 0;
+ grpc_byte_buffer_reader *reader = grpc_byte_buffer_reader_create(buffer);
+ gpr_slice next;
+ while (grpc_byte_buffer_reader_next(reader, &next) != 0){
+ memcpy(array + offset, GPR_SLICE_START_PTR(next), (size_t) GPR_SLICE_LENGTH(next));
+ offset += GPR_SLICE_LENGTH(next);
+ gpr_slice_unref(next);
+ }
+ grpc_byte_buffer_reader_destroy(reader);
+}
+
+static grpc_byte_buffer *CopyCharArrayToNewByteBuffer(const char *array, size_t length) {
+ gpr_slice slice = gpr_slice_from_copied_buffer(array, length);
+ grpc_byte_buffer *buffer = grpc_byte_buffer_create(&slice, 1);
+ gpr_slice_unref(slice);
+ return buffer;
+}
+
+@implementation NSData (GRPC)
++ (instancetype)grpc_dataWithByteBuffer:(grpc_byte_buffer *)buffer {
+ NSUInteger length = grpc_byte_buffer_length(buffer);
+ char *array = malloc(length * sizeof(*array));
+ if (!array) {
+ // TODO(jcanizales): grpc_byte_buffer is reference-counted, so we can
+ // prevent this memory problem by implementing a subclass of NSData
+ // that wraps the grpc_byte_buffer. Then enumerateByteRangesUsingBlock:
+ // can be implemented using a grpc_byte_buffer_reader.
+ return nil;
+ }
+ CopyByteBufferToCharArray(buffer, array);
+ return [self dataWithBytesNoCopy:array length:length freeWhenDone:YES];
+}
+
+- (grpc_byte_buffer *)grpc_byteBuffer {
+ // Some implementations of NSData, as well as grpc_byte_buffer, support O(1)
+ // appending of byte arrays by not using internally a single contiguous memory
+ // block for representation.
+ // The following implementation is thus not optimal, sometimes requiring two
+ // copies (one by self.bytes and another by gpr_slice_from_copied_buffer).
+ // If it turns out to be an issue, we can use enumerateByteRangesUsingblock:
+ // to create an array of gpr_slice objects to pass to grpc_byte_buffer_create.
+ // That would make it do exactly one copy, always.
+ return CopyCharArrayToNewByteBuffer((const char *)self.bytes, (size_t)self.length);
+}
+@end
diff --git a/src/objective-c/GRPCClient/private/NSDictionary+GRPC.h b/src/objective-c/GRPCClient/private/NSDictionary+GRPC.h
new file mode 100644
index 0000000..b717b10
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/NSDictionary+GRPC.h
@@ -0,0 +1,7 @@
+#import <Foundation/Foundation.h>
+
+struct grpc_metadata;
+
+@interface NSDictionary (GRPC)
++ (instancetype)grpc_dictionaryFromMetadata:(struct grpc_metadata *)entries count:(size_t)count;
+@end
diff --git a/src/objective-c/GRPCClient/private/NSDictionary+GRPC.m b/src/objective-c/GRPCClient/private/NSDictionary+GRPC.m
new file mode 100644
index 0000000..a24396d
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/NSDictionary+GRPC.m
@@ -0,0 +1,23 @@
+#import "NSDictionary+GRPC.h"
+
+#include <grpc.h>
+
+@implementation NSDictionary (GRPC)
++ (instancetype)grpc_dictionaryFromMetadata:(grpc_metadata *)entries count:(size_t)count {
+ NSMutableDictionary *metadata = [NSMutableDictionary dictionaryWithCapacity:count];
+ for (grpc_metadata *entry = entries; entry < entries + count; entry++) {
+ // TODO(jcanizales): Verify in a C library test that it's converting header names to lower case automatically.
+ NSString *name = [NSString stringWithUTF8String:entry->key];
+ if (!name) {
+ continue;
+ }
+ if (!metadata[name]) {
+ metadata[name] = [NSMutableArray array];
+ }
+ // TODO(jcanizales): Should we use a non-copy constructor?
+ [metadata[name] addObject:[NSData dataWithBytes:entry->value
+ length:entry->value_length]];
+ }
+ return metadata;
+}
+@end
diff --git a/src/objective-c/GRPCClient/private/NSError+GRPC.h b/src/objective-c/GRPCClient/private/NSError+GRPC.h
new file mode 100644
index 0000000..949d1dd
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/NSError+GRPC.h
@@ -0,0 +1,41 @@
+#import <Foundation/Foundation.h>
+
+// TODO(jcanizales): Make the domain string public.
+extern NSString *const kGRPCErrorDomain;
+
+// TODO(jcanizales): Make this public and document each code.
+typedef NS_ENUM(NSInteger, GRPCErrorCode) {
+ GRPCErrorCodeCancelled = 1,
+ GRPCErrorCodeUnknown = 2,
+ GRPCErrorCodeInvalidArgument = 3,
+ GRPCErrorCodeDeadlineExceeded = 4,
+ GRPCErrorCodeNotFound = 5,
+ GRPCErrorCodeAlreadyExists = 6,
+ GRPCErrorCodePermissionDenied = 7,
+ GRPCErrorCodeUnauthenticated = 16,
+ GRPCErrorCodeResourceExhausted = 8,
+ GRPCErrorCodeFailedPrecondition = 9,
+ GRPCErrorCodeAborted = 10,
+ GRPCErrorCodeOutOfRange = 11,
+ GRPCErrorCodeUnimplemented = 12,
+ GRPCErrorCodeInternal = 13,
+ GRPCErrorCodeUnavailable = 14,
+ GRPCErrorCodeDataLoss = 15
+};
+
+// TODO(jcanizales): This is conflating trailing metadata with Status details. Fix it once there's
+// a decision on how to codify Status.
+#include <grpc/status.h>
+struct grpc_metadata;
+struct grpc_status {
+ grpc_status_code status;
+ const char *details;
+ size_t metadata_count;
+ struct grpc_metadata *metadata_elements;
+};
+
+@interface NSError (GRPC)
+// Returns nil if the status is OK. Otherwise, a NSError whose code is one of
+// GRPCErrorCode and whose domain is kGRPCErrorDomain.
++ (instancetype)grpc_errorFromStatus:(struct grpc_status *)status;
+@end
diff --git a/src/objective-c/GRPCClient/private/NSError+GRPC.m b/src/objective-c/GRPCClient/private/NSError+GRPC.m
new file mode 100644
index 0000000..73ce112
--- /dev/null
+++ b/src/objective-c/GRPCClient/private/NSError+GRPC.m
@@ -0,0 +1,18 @@
+#import "NSError+GRPC.h"
+
+#include <grpc.h>
+
+NSString *const kGRPCErrorDomain = @"org.grpc";
+
+@implementation NSError (GRPC)
++ (instancetype)grpc_errorFromStatus:(struct grpc_status *)status {
+ if (status->status == GRPC_STATUS_OK) {
+ return nil;
+ }
+ NSString *message =
+ [NSString stringWithFormat:@"Code=%i Message='%s'", status->status, status->details];
+ return [NSError errorWithDomain:kGRPCErrorDomain
+ code:status->status
+ userInfo:@{NSLocalizedDescriptionKey: message}];
+}
+@end