|
|
|
@ -250,11 +250,13 @@ static NSString * const kBearerPrefix = @"Bearer "; |
|
|
|
|
if (self.state == GRXWriterStatePaused) { |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
__weak GRXConcurrentWriteable *weakWriteable = _responseWriteable; |
|
|
|
|
|
|
|
|
|
dispatch_async(_callQueue, ^{ |
|
|
|
|
[weakSelf startReadWithHandler:^(grpc_byte_buffer *message) { |
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
__weak GRXConcurrentWriteable *weakWriteable = self->_responseWriteable; |
|
|
|
|
[self startReadWithHandler:^(grpc_byte_buffer *message) { |
|
|
|
|
__strong GRPCCall *strongSelf = weakSelf; |
|
|
|
|
__strong GRXConcurrentWriteable *strongWriteable = weakWriteable; |
|
|
|
|
if (message == NULL) { |
|
|
|
|
// No more messages from the server |
|
|
|
|
return; |
|
|
|
@ -266,14 +268,14 @@ static NSString * const kBearerPrefix = @"Bearer "; |
|
|
|
|
// 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. |
|
|
|
|
[weakSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain |
|
|
|
|
code:GRPCErrorCodeResourceExhausted |
|
|
|
|
userInfo:@{NSLocalizedDescriptionKey: @"Client does not have enough memory to hold the server response."}]]; |
|
|
|
|
[weakSelf cancelCall]; |
|
|
|
|
[strongSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain |
|
|
|
|
code:GRPCErrorCodeResourceExhausted |
|
|
|
|
userInfo:@{NSLocalizedDescriptionKey: @"Client does not have enough memory to hold the server response."}]]; |
|
|
|
|
[strongSelf cancelCall]; |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
[weakWriteable enqueueValue:data completionHandler:^{ |
|
|
|
|
[weakSelf startNextRead]; |
|
|
|
|
[strongWriteable enqueueValue:data completionHandler:^{ |
|
|
|
|
[strongSelf startNextRead]; |
|
|
|
|
}]; |
|
|
|
|
}]; |
|
|
|
|
}); |
|
|
|
@ -333,12 +335,15 @@ static NSString * const kBearerPrefix = @"Bearer "; |
|
|
|
|
_requestWriter.state = GRXWriterStatePaused; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
dispatch_async(_callQueue, ^{ |
|
|
|
|
[weakSelf writeMessage:value withErrorHandler:^{ |
|
|
|
|
[weakSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain |
|
|
|
|
code:GRPCErrorCodeInternal |
|
|
|
|
userInfo:nil]]; |
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
[self writeMessage:value withErrorHandler:^{ |
|
|
|
|
__strong GRPCCall *strongSelf = weakSelf; |
|
|
|
|
if (strongSelf != nil) { |
|
|
|
|
[strongSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain |
|
|
|
|
code:GRPCErrorCodeInternal |
|
|
|
|
userInfo:nil]]; |
|
|
|
|
} |
|
|
|
|
}]; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
@ -360,12 +365,13 @@ static NSString * const kBearerPrefix = @"Bearer "; |
|
|
|
|
if (errorOrNil) { |
|
|
|
|
[self cancel]; |
|
|
|
|
} else { |
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
dispatch_async(_callQueue, ^{ |
|
|
|
|
[weakSelf finishRequestWithErrorHandler:^{ |
|
|
|
|
[weakSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain |
|
|
|
|
code:GRPCErrorCodeInternal |
|
|
|
|
userInfo:nil]]; |
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
[self finishRequestWithErrorHandler:^{ |
|
|
|
|
__strong GRPCCall *strongSelf = weakSelf; |
|
|
|
|
[strongSelf finishWithError:[NSError errorWithDomain:kGRPCErrorDomain |
|
|
|
|
code:GRPCErrorCodeInternal |
|
|
|
|
userInfo:nil]]; |
|
|
|
|
}]; |
|
|
|
|
}); |
|
|
|
|
} |
|
|
|
@ -387,30 +393,37 @@ static NSString * const kBearerPrefix = @"Bearer "; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
- (void)invokeCall { |
|
|
|
|
__weak GRPCCall *weakSelf = self; |
|
|
|
|
[self invokeCallWithHeadersHandler:^(NSDictionary *headers) { |
|
|
|
|
// Response headers received. |
|
|
|
|
self.responseHeaders = headers; |
|
|
|
|
[self startNextRead]; |
|
|
|
|
__strong GRPCCall *strongSelf = weakSelf; |
|
|
|
|
if (strongSelf) { |
|
|
|
|
strongSelf.responseHeaders = headers; |
|
|
|
|
[strongSelf startNextRead]; |
|
|
|
|
} |
|
|
|
|
} completionHandler:^(NSError *error, NSDictionary *trailers) { |
|
|
|
|
self.responseTrailers = trailers; |
|
|
|
|
__strong GRPCCall *strongSelf = weakSelf; |
|
|
|
|
if (strongSelf) { |
|
|
|
|
strongSelf.responseTrailers = trailers; |
|
|
|
|
|
|
|
|
|
if (error) { |
|
|
|
|
NSMutableDictionary *userInfo = [NSMutableDictionary dictionary]; |
|
|
|
|
if (error.userInfo) { |
|
|
|
|
[userInfo addEntriesFromDictionary:error.userInfo]; |
|
|
|
|
} |
|
|
|
|
userInfo[kGRPCTrailersKey] = self.responseTrailers; |
|
|
|
|
// TODO(jcanizales): The C gRPC library doesn't guarantee that the headers block will be |
|
|
|
|
// called before this one, so an error might end up with trailers but no headers. We |
|
|
|
|
// shouldn't call finishWithError until ater both blocks are called. It is also when this is |
|
|
|
|
// done that we can provide a merged view of response headers and trailers in a thread-safe |
|
|
|
|
// way. |
|
|
|
|
if (self.responseHeaders) { |
|
|
|
|
userInfo[kGRPCHeadersKey] = self.responseHeaders; |
|
|
|
|
if (error) { |
|
|
|
|
NSMutableDictionary *userInfo = [NSMutableDictionary dictionary]; |
|
|
|
|
if (error.userInfo) { |
|
|
|
|
[userInfo addEntriesFromDictionary:error.userInfo]; |
|
|
|
|
} |
|
|
|
|
userInfo[kGRPCTrailersKey] = strongSelf.responseTrailers; |
|
|
|
|
// TODO(jcanizales): The C gRPC library doesn't guarantee that the headers block will be |
|
|
|
|
// called before this one, so an error might end up with trailers but no headers. We |
|
|
|
|
// shouldn't call finishWithError until ater both blocks are called. It is also when this is |
|
|
|
|
// done that we can provide a merged view of response headers and trailers in a thread-safe |
|
|
|
|
// way. |
|
|
|
|
if (strongSelf.responseHeaders) { |
|
|
|
|
userInfo[kGRPCHeadersKey] = strongSelf.responseHeaders; |
|
|
|
|
} |
|
|
|
|
error = [NSError errorWithDomain:error.domain code:error.code userInfo:userInfo]; |
|
|
|
|
} |
|
|
|
|
error = [NSError errorWithDomain:error.domain code:error.code userInfo:userInfo]; |
|
|
|
|
[strongSelf finishWithError:error]; |
|
|
|
|
} |
|
|
|
|
[self finishWithError:error]; |
|
|
|
|
}]; |
|
|
|
|
// Now that the RPC has been initiated, request writes can start. |
|
|
|
|
@synchronized(_requestWriter) { |
|
|
|
|