Proto-related changes

4 files changed, 278 insertions, 3 deletions

diff --git a/src/objective-c/ProtoRPC/ProtoRPC.h b/src/objective-c/ProtoRPC/ProtoRPC.h
index 45d3526092..1a27cac2a3 100644
--- a/src/objective-c/ProtoRPC/ProtoRPC.h
+++ b/src/objective-c/ProtoRPC/ProtoRPC.h
@@ -21,6 +21,55 @@
 
 #import "ProtoMethod.h"
 
+@class GPBMessage;
+
+/** A unary-request RPC call with Protobuf. */
+@interface GRPCUnaryProtoCall : NSObject
+
+/**
+ * Users should not use this initializer directly. Call objects will be created, initialized, and
+ * returned to users by methods of the generated service.
+ */
+- (instancetype)initWithRequestOptions:(GRPCRequestOptions *)requestOptions
+                               message:(GPBMessage *)message
+                       responseHandler:(id<GRPCResponseHandler>)handler
+                           callOptions:(GRPCCallOptions *)callOptions
+                         responseClass:(Class)responseClass;
+
+/** Cancel the call at best effort. */
+- (void)cancel;
+
+@end
+
+/** A client-streaming RPC call with Protobuf. */
+@interface GRPCStreamingProtoCall : NSObject
+
+/**
+ * Users should not use this initializer directly. Call objects will be created, initialized, and
+ * returned to users by methods of the generated service.
+ */
+- (instancetype)initWithRequestOptions:(GRPCRequestOptions *)requestOptions
+                       responseHandler:(id<GRPCResponseHandler>)handler
+                           callOptions:(GRPCCallOptions *)callOptions
+                         responseClass:(Class)responseClass;
+
+/** Cancel the call at best effort. */
+- (void)cancel;
+
+/**
+ * Send a message to the server. The message should be a Protobuf message which will be serialized
+ * internally.
+ */
+- (void)writeWithMessage:(GPBMessage *)message;
+
+/**
+ * Finish the RPC request and half-close the call. The server may still send messages and/or
+ * trailers to the client.
+ */
+- (void)finish;
+
+@end
+
 __attribute__((deprecated("Please use GRPCProtoCall."))) @interface ProtoRPC
     : GRPCCall
 
diff --git a/src/objective-c/ProtoRPC/ProtoRPC.m b/src/objective-c/ProtoRPC/ProtoRPC.m
index 5dca971b08..f89c160650 100644
--- a/src/objective-c/ProtoRPC/ProtoRPC.m
+++ b/src/objective-c/ProtoRPC/ProtoRPC.m
@@ -23,9 +23,166 @@
 #else
 #import <GPBProtocolBuffers.h>
 #endif
+#import <GRPCClient/GRPCCall.h>
 #import <RxLibrary/GRXWriteable.h>
 #import <RxLibrary/GRXWriter+Transformations.h>
 
+@implementation GRPCUnaryProtoCall {
+  GRPCStreamingProtoCall *_call;
+}
+
+- (instancetype)initWithRequestOptions:(GRPCRequestOptions *)requestOptions
+                               message:(GPBMessage *)message
+                       responseHandler:(id<GRPCResponseHandler>)handler
+                           callOptions:(GRPCCallOptions *)callOptions
+                         responseClass:(Class)responseClass {
+  if ((self = [super init])) {
+    _call = [[GRPCStreamingProtoCall alloc] initWithRequestOptions:requestOptions
+                                                   responseHandler:handler
+                                                       callOptions:callOptions
+                                                     responseClass:responseClass];
+    [_call writeWithMessage:message];
+    [_call finish];
+  }
+  return self;
+}
+
+- (void)cancel {
+  [_call cancel];
+  _call = nil;
+}
+
+@end
+
+@interface GRPCStreamingProtoCall ()<GRPCResponseHandler>
+
+@end
+
+@implementation GRPCStreamingProtoCall {
+  GRPCRequestOptions *_requestOptions;
+  id<GRPCResponseHandler> _handler;
+  GRPCCallOptions *_callOptions;
+  Class _responseClass;
+
+  GRPCCall2 *_call;
+  dispatch_queue_t _dispatchQueue;
+}
+
+- (instancetype)initWithRequestOptions:(GRPCRequestOptions *)requestOptions
+                       responseHandler:(id<GRPCResponseHandler>)handler
+                           callOptions:(GRPCCallOptions *)callOptions
+                         responseClass:(Class)responseClass {
+  if ((self = [super init])) {
+    _requestOptions = [requestOptions copy];
+    _handler = handler;
+    _callOptions = [callOptions copy];
+    _responseClass = responseClass;
+    _dispatchQueue = dispatch_queue_create(nil, DISPATCH_QUEUE_SERIAL);
+
+    [self start];
+  }
+  return self;
+}
+
+- (void)start {
+  _call = [[GRPCCall2 alloc] initWithRequestOptions:_requestOptions
+                                            handler:self
+                                        callOptions:_callOptions];
+  [_call start];
+}
+
+- (void)cancel {
+  dispatch_async(_dispatchQueue, ^{
+    if (_call) {
+      [_call cancel];
+      _call = nil;
+    }
+    if (_handler) {
+      id<GRPCResponseHandler> handler = _handler;
+      dispatch_async(handler.dispatchQueue, ^{
+        [handler closedWithTrailingMetadata:nil
+                                      error:[NSError errorWithDomain:kGRPCErrorDomain
+                                                                code:GRPCErrorCodeCancelled
+                                                            userInfo:@{
+                                                              NSLocalizedDescriptionKey :
+                                                                  @"Canceled by app"
+                                                            }]];
+      });
+      _handler = nil;
+    }
+  });
+}
+
+- (void)writeWithMessage:(GPBMessage *)message {
+  if (![message isKindOfClass:[GPBMessage class]]) {
+    [NSException raise:NSInvalidArgumentException format:@"Data must be a valid protobuf type."];
+  }
+
+  dispatch_async(_dispatchQueue, ^{
+    if (_call) {
+      [_call writeWithData:[message data]];
+    }
+  });
+}
+
+- (void)finish {
+  dispatch_async(_dispatchQueue, ^{
+    if (_call) {
+      [_call finish];
+      _call = nil;
+    }
+  });
+}
+
+- (void)receivedInitialMetadata:(NSDictionary *)initialMetadata {
+  if (_handler) {
+    id<GRPCResponseHandler> handler = _handler;
+    dispatch_async(handler.dispatchQueue, ^{
+      [handler receivedInitialMetadata:initialMetadata];
+    });
+  }
+}
+
+- (void)receivedMessage:(NSData *)message {
+  if (_handler) {
+    id<GRPCResponseHandler> handler = _handler;
+    NSError *error = nil;
+    id parsed = [_responseClass parseFromData:message error:&error];
+    if (parsed) {
+      dispatch_async(handler.dispatchQueue, ^{
+        [handler receivedMessage:parsed];
+      });
+    } else {
+      dispatch_async(handler.dispatchQueue, ^{
+        [handler closedWithTrailingMetadata:nil error:error];
+      });
+      handler = nil;
+      [_call cancel];
+      _call = nil;
+    }
+  }
+}
+
+- (void)closedWithTrailingMetadata:(NSDictionary *)trailingMetadata error:(NSError *)error {
+  if (_handler) {
+    id<GRPCResponseHandler> handler = _handler;
+    dispatch_async(handler.dispatchQueue, ^{
+      [handler closedWithTrailingMetadata:trailingMetadata error:error];
+    });
+    _handler = nil;
+  }
+  if (_call) {
+    [_call cancel];
+    _call = nil;
+  }
+}
+
+- (dispatch_queue_t)dispatchQueue {
+  return _dispatchQueue;
+}
+
+@end
+
 static NSError *ErrorForBadProto(id proto, Class expectedClass, NSError *parsingError) {
   NSDictionary *info = @{
     NSLocalizedDescriptionKey : @"Unable to parse response from the server",
diff --git a/src/objective-c/ProtoRPC/ProtoService.h b/src/objective-c/ProtoRPC/ProtoService.h
index 29c4e9be36..2105de78a3 100644
--- a/src/objective-c/ProtoRPC/ProtoService.h
+++ b/src/objective-c/ProtoRPC/ProtoService.h
@@ -21,16 +21,40 @@
 @class GRPCProtoCall;
 @protocol GRXWriteable;
 @class GRXWriter;
+@class GRPCCallOptions;
+@class GRPCProtoCall;
+@class GRPCUnaryProtoCall;
+@class GRPCStreamingProtoCall;
+@protocol GRPCProtoResponseCallbacks;
 
 __attribute__((deprecated("Please use GRPCProtoService."))) @interface ProtoService
-    : NSObject -
+    : NSObject
+
+      -
       (instancetype)initWithHost : (NSString *)host packageName
-    : (NSString *)packageName serviceName : (NSString *)serviceName NS_DESIGNATED_INITIALIZER;
+    : (NSString *)packageName serviceName : (NSString *)serviceName callOptions
+    : (GRPCCallOptions *)callOptions NS_DESIGNATED_INITIALIZER;
+
+- (instancetype)initWithHost:(NSString *)host
+                 packageName:(NSString *)packageName
+                 serviceName:(NSString *)serviceName;
 
 - (GRPCProtoCall *)RPCToMethod:(NSString *)method
                 requestsWriter:(GRXWriter *)requestsWriter
                  responseClass:(Class)responseClass
             responsesWriteable:(id<GRXWriteable>)responsesWriteable;
+
+- (GRPCUnaryProtoCall *)RPCToMethod:(NSString *)method
+                            message:(id)message
+                    responseHandler:(id<GRPCProtoResponseCallbacks>)handler
+                        callOptions:(GRPCCallOptions *)callOptions
+                      responseClass:(Class)responseClass;
+
+- (GRPCStreamingProtoCall *)RPCToMethod:(NSString *)method
+                        responseHandler:(id<GRPCProtoResponseCallbacks>)handler
+                            callOptions:(GRPCCallOptions *)callOptions
+                          responseClass:(Class)responseClass;
+
 @end
 
 /**
diff --git a/src/objective-c/ProtoRPC/ProtoService.m b/src/objective-c/ProtoRPC/ProtoService.m
index 3e9cc5c0b2..6df502fb0c 100644
--- a/src/objective-c/ProtoRPC/ProtoService.m
+++ b/src/objective-c/ProtoRPC/ProtoService.m
@@ -18,6 +18,7 @@
 
 #import "ProtoService.h"
 
+#import <GRPCClient/GRPCCall.h>
 #import <RxLibrary/GRXWriteable.h>
 #import <RxLibrary/GRXWriter.h>
 
@@ -31,6 +32,7 @@
   NSString *_host;
   NSString *_packageName;
   NSString *_serviceName;
+  GRPCCallOptions *_callOptions;
 }
 
 - (instancetype)init {
@@ -40,7 +42,8 @@
 // Designated initializer
 - (instancetype)initWithHost:(NSString *)host
                  packageName:(NSString *)packageName
-                 serviceName:(NSString *)serviceName {
+                 serviceName:(NSString *)serviceName
+                 callOptions:(GRPCCallOptions *)callOptions {
   if (!host || !serviceName) {
     [NSException raise:NSInvalidArgumentException
                 format:@"Neither host nor serviceName can be nil."];
@@ -49,10 +52,17 @@
     _host = [host copy];
     _packageName = [packageName copy];
     _serviceName = [serviceName copy];
+    _callOptions = [callOptions copy];
   }
   return self;
 }
 
+- (instancetype)initWithHost:(NSString *)host
+                 packageName:(NSString *)packageName
+                 serviceName:(NSString *)serviceName {
+  return [self initWithHost:host packageName:packageName serviceName:serviceName callOptions:nil];
+}
+
 - (GRPCProtoCall *)RPCToMethod:(NSString *)method
                 requestsWriter:(GRXWriter *)requestsWriter
                  responseClass:(Class)responseClass
@@ -65,6 +75,41 @@
                                responseClass:responseClass
                           responsesWriteable:responsesWriteable];
 }
+
+- (GRPCUnaryProtoCall *)RPCToMethod:(NSString *)method
+                            message:(id)message
+                    responseHandler:(id<GRPCProtoResponseCallbacks>)handler
+                        callOptions:(GRPCCallOptions *)callOptions
+                      responseClass:(Class)responseClass {
+  GRPCProtoMethod *methodName =
+      [[GRPCProtoMethod alloc] initWithPackage:_packageName service:_serviceName method:method];
+  GRPCRequestOptions *requestOptions =
+      [[GRPCRequestOptions alloc] initWithHost:_host
+                                          path:methodName.HTTPPath
+                                        safety:GRPCCallSafetyDefault];
+  return [[GRPCUnaryProtoCall alloc] initWithRequestOptions:requestOptions
+                                                    message:message
+                                            responseHandler:handler
+                                                callOptions:callOptions ?: _callOptions
+                                              responseClass:responseClass];
+}
+
+- (GRPCStreamingProtoCall *)RPCToMethod:(NSString *)method
+                        responseHandler:(id<GRPCProtoResponseCallbacks>)handler
+                            callOptions:(GRPCCallOptions *)callOptions
+                          responseClass:(Class)responseClass {
+  GRPCProtoMethod *methodName =
+      [[GRPCProtoMethod alloc] initWithPackage:_packageName service:_serviceName method:method];
+  GRPCRequestOptions *requestOptions =
+      [[GRPCRequestOptions alloc] initWithHost:_host
+                                          path:methodName.HTTPPath
+                                        safety:GRPCCallSafetyDefault];
+  return [[GRPCStreamingProtoCall alloc] initWithRequestOptions:requestOptions
+                                                responseHandler:handler
+                                                    callOptions:callOptions ?: _callOptions
+                                                  responseClass:responseClass];
+}
+
 @end
 
 @implementation GRPCProtoService