src/objective-c/GRPCClient/GRPCCall.m

#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