• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1/*
2 *
3 * Copyright 2015 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 *     http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19#import "InteropTests.h"
20
21#include <grpc/status.h>
22
23#import <GRPCClient/GRPCCall+ChannelArg.h>
24#import <GRPCClient/GRPCCall+Cronet.h>
25#import <GRPCClient/GRPCCall+Interceptor.h>
26#import <GRPCClient/GRPCCall+Tests.h>
27#import <GRPCClient/GRPCInterceptor.h>
28#import <GRPCClient/internal_testing/GRPCCall+InternalTests.h>
29#import <ProtoRPC/ProtoRPC.h>
30#import <RxLibrary/GRXBufferedPipe.h>
31#import <RxLibrary/GRXWriter+Immediate.h>
32#import <grpc/grpc.h>
33#import <grpc/support/log.h>
34#import "src/objective-c/tests/RemoteTestClient/Messages.pbobjc.h"
35#import "src/objective-c/tests/RemoteTestClient/Test.pbobjc.h"
36#import "src/objective-c/tests/RemoteTestClient/Test.pbrpc.h"
37
38#import "InteropTestsBlockCallbacks.h"
39
40#define TEST_TIMEOUT 32
41
42static const int kTestRetries = 3;
43extern const char *kCFStreamVarName;
44
45// Convenience constructors for the generated proto messages:
46
47@interface RMTStreamingOutputCallRequest (Constructors)
48+ (instancetype)messageWithPayloadSize:(NSNumber *)payloadSize
49                 requestedResponseSize:(NSNumber *)responseSize;
50@end
51
52@implementation RMTStreamingOutputCallRequest (Constructors)
53+ (instancetype)messageWithPayloadSize:(NSNumber *)payloadSize
54                 requestedResponseSize:(NSNumber *)responseSize {
55  RMTStreamingOutputCallRequest *request = [self message];
56  RMTResponseParameters *parameters = [RMTResponseParameters message];
57  parameters.size = responseSize.intValue;
58  [request.responseParametersArray addObject:parameters];
59  request.payload.body = [NSMutableData dataWithLength:payloadSize.unsignedIntegerValue];
60  return request;
61}
62@end
63
64@interface RMTStreamingOutputCallResponse (Constructors)
65+ (instancetype)messageWithPayloadSize:(NSNumber *)payloadSize;
66@end
67
68@implementation RMTStreamingOutputCallResponse (Constructors)
69+ (instancetype)messageWithPayloadSize:(NSNumber *)payloadSize {
70  RMTStreamingOutputCallResponse *response = [self message];
71  response.payload.type = RMTPayloadType_Compressable;
72  response.payload.body = [NSMutableData dataWithLength:payloadSize.unsignedIntegerValue];
73  return response;
74}
75@end
76
77BOOL isRemoteInteropTest(NSString *host) {
78  return [host isEqualToString:@"grpc-test.sandbox.googleapis.com"];
79}
80
81@interface DefaultInterceptorFactory : NSObject <GRPCInterceptorFactory>
82
83- (GRPCInterceptor *)createInterceptorWithManager:(GRPCInterceptorManager *)interceptorManager;
84
85@end
86
87@implementation DefaultInterceptorFactory
88
89- (GRPCInterceptor *)createInterceptorWithManager:(GRPCInterceptorManager *)interceptorManager {
90  dispatch_queue_t queue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
91  return [[GRPCInterceptor alloc] initWithInterceptorManager:interceptorManager
92                                               dispatchQueue:queue];
93}
94
95@end
96
97@interface HookInterceptorFactory : NSObject <GRPCInterceptorFactory>
98
99- (instancetype)
100      initWithDispatchQueue:(dispatch_queue_t)dispatchQueue
101                  startHook:(void (^)(GRPCRequestOptions *requestOptions,
102                                      GRPCCallOptions *callOptions,
103                                      GRPCInterceptorManager *manager))startHook
104              writeDataHook:(void (^)(id data, GRPCInterceptorManager *manager))writeDataHook
105                 finishHook:(void (^)(GRPCInterceptorManager *manager))finishHook
106    receiveNextMessagesHook:(void (^)(NSUInteger numberOfMessages,
107                                      GRPCInterceptorManager *manager))receiveNextMessagesHook
108         responseHeaderHook:(void (^)(NSDictionary *initialMetadata,
109                                      GRPCInterceptorManager *manager))responseHeaderHook
110           responseDataHook:(void (^)(id data, GRPCInterceptorManager *manager))responseDataHook
111          responseCloseHook:(void (^)(NSDictionary *trailingMetadata, NSError *error,
112                                      GRPCInterceptorManager *manager))responseCloseHook
113           didWriteDataHook:(void (^)(GRPCInterceptorManager *manager))didWriteDataHook;
114
115- (GRPCInterceptor *)createInterceptorWithManager:(GRPCInterceptorManager *)interceptorManager;
116
117@end
118
119@interface HookInterceptor : GRPCInterceptor
120
121- (instancetype)
122    initWithInterceptorManager:(GRPCInterceptorManager *)interceptorManager
123                 dispatchQueue:(dispatch_queue_t)dispatchQueue
124                     startHook:(void (^)(GRPCRequestOptions *requestOptions,
125                                         GRPCCallOptions *callOptions,
126                                         GRPCInterceptorManager *manager))startHook
127                 writeDataHook:(void (^)(id data, GRPCInterceptorManager *manager))writeDataHook
128                    finishHook:(void (^)(GRPCInterceptorManager *manager))finishHook
129       receiveNextMessagesHook:(void (^)(NSUInteger numberOfMessages,
130                                         GRPCInterceptorManager *manager))receiveNextMessagesHook
131            responseHeaderHook:(void (^)(NSDictionary *initialMetadata,
132                                         GRPCInterceptorManager *manager))responseHeaderHook
133              responseDataHook:(void (^)(id data, GRPCInterceptorManager *manager))responseDataHook
134             responseCloseHook:(void (^)(NSDictionary *trailingMetadata, NSError *error,
135                                         GRPCInterceptorManager *manager))responseCloseHook
136              didWriteDataHook:(void (^)(GRPCInterceptorManager *manager))didWriteDataHook;
137
138@end
139
140@implementation HookInterceptorFactory {
141 @protected
142  void (^_startHook)(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
143                     GRPCInterceptorManager *manager);
144  void (^_writeDataHook)(id data, GRPCInterceptorManager *manager);
145  void (^_finishHook)(GRPCInterceptorManager *manager);
146  void (^_receiveNextMessagesHook)(NSUInteger numberOfMessages, GRPCInterceptorManager *manager);
147  void (^_responseHeaderHook)(NSDictionary *initialMetadata, GRPCInterceptorManager *manager);
148  void (^_responseDataHook)(id data, GRPCInterceptorManager *manager);
149  void (^_responseCloseHook)(NSDictionary *trailingMetadata, NSError *error,
150                             GRPCInterceptorManager *manager);
151  void (^_didWriteDataHook)(GRPCInterceptorManager *manager);
152  dispatch_queue_t _dispatchQueue;
153}
154
155- (instancetype)
156      initWithDispatchQueue:(dispatch_queue_t)dispatchQueue
157                  startHook:(void (^)(GRPCRequestOptions *requestOptions,
158                                      GRPCCallOptions *callOptions,
159                                      GRPCInterceptorManager *manager))startHook
160              writeDataHook:(void (^)(id data, GRPCInterceptorManager *manager))writeDataHook
161                 finishHook:(void (^)(GRPCInterceptorManager *manager))finishHook
162    receiveNextMessagesHook:(void (^)(NSUInteger numberOfMessages,
163                                      GRPCInterceptorManager *manager))receiveNextMessagesHook
164         responseHeaderHook:(void (^)(NSDictionary *initialMetadata,
165                                      GRPCInterceptorManager *manager))responseHeaderHook
166           responseDataHook:(void (^)(id data, GRPCInterceptorManager *manager))responseDataHook
167          responseCloseHook:(void (^)(NSDictionary *trailingMetadata, NSError *error,
168                                      GRPCInterceptorManager *manager))responseCloseHook
169           didWriteDataHook:(void (^)(GRPCInterceptorManager *manager))didWriteDataHook {
170  if ((self = [super init])) {
171    _dispatchQueue = dispatchQueue;
172    _startHook = startHook;
173    _writeDataHook = writeDataHook;
174    _finishHook = finishHook;
175    _receiveNextMessagesHook = receiveNextMessagesHook;
176    _responseHeaderHook = responseHeaderHook;
177    _responseDataHook = responseDataHook;
178    _responseCloseHook = responseCloseHook;
179    _didWriteDataHook = didWriteDataHook;
180  }
181  return self;
182}
183
184- (GRPCInterceptor *)createInterceptorWithManager:(GRPCInterceptorManager *)interceptorManager {
185  return [[HookInterceptor alloc] initWithInterceptorManager:interceptorManager
186                                               dispatchQueue:_dispatchQueue
187                                                   startHook:_startHook
188                                               writeDataHook:_writeDataHook
189                                                  finishHook:_finishHook
190                                     receiveNextMessagesHook:_receiveNextMessagesHook
191                                          responseHeaderHook:_responseHeaderHook
192                                            responseDataHook:_responseDataHook
193                                           responseCloseHook:_responseCloseHook
194                                            didWriteDataHook:_didWriteDataHook];
195}
196
197@end
198
199@implementation HookInterceptor {
200  void (^_startHook)(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
201                     GRPCInterceptorManager *manager);
202  void (^_writeDataHook)(id data, GRPCInterceptorManager *manager);
203  void (^_finishHook)(GRPCInterceptorManager *manager);
204  void (^_receiveNextMessagesHook)(NSUInteger numberOfMessages, GRPCInterceptorManager *manager);
205  void (^_responseHeaderHook)(NSDictionary *initialMetadata, GRPCInterceptorManager *manager);
206  void (^_responseDataHook)(id data, GRPCInterceptorManager *manager);
207  void (^_responseCloseHook)(NSDictionary *trailingMetadata, NSError *error,
208                             GRPCInterceptorManager *manager);
209  void (^_didWriteDataHook)(GRPCInterceptorManager *manager);
210  GRPCInterceptorManager *_manager;
211  dispatch_queue_t _dispatchQueue;
212}
213
214- (dispatch_queue_t)dispatchQueue {
215  return _dispatchQueue;
216}
217
218- (instancetype)
219    initWithInterceptorManager:(GRPCInterceptorManager *)interceptorManager
220                 dispatchQueue:(dispatch_queue_t)dispatchQueue
221                     startHook:(void (^)(GRPCRequestOptions *requestOptions,
222                                         GRPCCallOptions *callOptions,
223                                         GRPCInterceptorManager *manager))startHook
224                 writeDataHook:(void (^)(id data, GRPCInterceptorManager *manager))writeDataHook
225                    finishHook:(void (^)(GRPCInterceptorManager *manager))finishHook
226       receiveNextMessagesHook:(void (^)(NSUInteger numberOfMessages,
227                                         GRPCInterceptorManager *manager))receiveNextMessagesHook
228            responseHeaderHook:(void (^)(NSDictionary *initialMetadata,
229                                         GRPCInterceptorManager *manager))responseHeaderHook
230              responseDataHook:(void (^)(id data, GRPCInterceptorManager *manager))responseDataHook
231             responseCloseHook:(void (^)(NSDictionary *trailingMetadata, NSError *error,
232                                         GRPCInterceptorManager *manager))responseCloseHook
233              didWriteDataHook:(void (^)(GRPCInterceptorManager *manager))didWriteDataHook {
234  if ((self = [super initWithInterceptorManager:interceptorManager dispatchQueue:dispatchQueue])) {
235    _startHook = startHook;
236    _writeDataHook = writeDataHook;
237    _finishHook = finishHook;
238    _receiveNextMessagesHook = receiveNextMessagesHook;
239    _responseHeaderHook = responseHeaderHook;
240    _responseDataHook = responseDataHook;
241    _responseCloseHook = responseCloseHook;
242    _didWriteDataHook = didWriteDataHook;
243    _dispatchQueue = dispatchQueue;
244    _manager = interceptorManager;
245  }
246  return self;
247}
248
249- (void)startWithRequestOptions:(GRPCRequestOptions *)requestOptions
250                    callOptions:(GRPCCallOptions *)callOptions {
251  if (_startHook) {
252    _startHook(requestOptions, callOptions, _manager);
253  }
254}
255
256- (void)writeData:(id)data {
257  if (_writeDataHook) {
258    _writeDataHook(data, _manager);
259  }
260}
261
262- (void)finish {
263  if (_finishHook) {
264    _finishHook(_manager);
265  }
266}
267
268- (void)receiveNextMessages:(NSUInteger)numberOfMessages {
269  if (_receiveNextMessagesHook) {
270    _receiveNextMessagesHook(numberOfMessages, _manager);
271  }
272}
273
274- (void)didReceiveInitialMetadata:(NSDictionary *)initialMetadata {
275  if (_responseHeaderHook) {
276    _responseHeaderHook(initialMetadata, _manager);
277  }
278}
279
280- (void)didReceiveData:(id)data {
281  if (_responseDataHook) {
282    _responseDataHook(data, _manager);
283  }
284}
285
286- (void)didCloseWithTrailingMetadata:(NSDictionary *)trailingMetadata error:(NSError *)error {
287  if (_responseCloseHook) {
288    _responseCloseHook(trailingMetadata, error, _manager);
289  }
290}
291
292- (void)didWriteData {
293  if (_didWriteDataHook) {
294    _didWriteDataHook(_manager);
295  }
296}
297
298@end
299
300@interface GlobalInterceptorFactory : HookInterceptorFactory
301
302@property BOOL enabled;
303
304- (instancetype)initWithDispatchQueue:(dispatch_queue_t)dispatchQueue;
305
306- (void)setStartHook:(void (^)(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
307                               GRPCInterceptorManager *manager))startHook
308              writeDataHook:(void (^)(id data, GRPCInterceptorManager *manager))writeDataHook
309                 finishHook:(void (^)(GRPCInterceptorManager *manager))finishHook
310    receiveNextMessagesHook:(void (^)(NSUInteger numberOfMessages,
311                                      GRPCInterceptorManager *manager))receiveNextMessagesHook
312         responseHeaderHook:(void (^)(NSDictionary *initialMetadata,
313                                      GRPCInterceptorManager *manager))responseHeaderHook
314           responseDataHook:(void (^)(id data, GRPCInterceptorManager *manager))responseDataHook
315          responseCloseHook:(void (^)(NSDictionary *trailingMetadata, NSError *error,
316                                      GRPCInterceptorManager *manager))responseCloseHook
317           didWriteDataHook:(void (^)(GRPCInterceptorManager *manager))didWriteDataHook;
318
319@end
320
321@implementation GlobalInterceptorFactory
322
323- (instancetype)initWithDispatchQueue:(dispatch_queue_t)dispatchQueue {
324  _enabled = NO;
325  return [super initWithDispatchQueue:dispatchQueue
326                            startHook:nil
327                        writeDataHook:nil
328                           finishHook:nil
329              receiveNextMessagesHook:nil
330                   responseHeaderHook:nil
331                     responseDataHook:nil
332                    responseCloseHook:nil
333                     didWriteDataHook:nil];
334}
335
336- (GRPCInterceptor *)createInterceptorWithManager:(GRPCInterceptorManager *)interceptorManager {
337  if (_enabled) {
338    return [[HookInterceptor alloc] initWithInterceptorManager:interceptorManager
339                                                 dispatchQueue:_dispatchQueue
340                                                     startHook:_startHook
341                                                 writeDataHook:_writeDataHook
342                                                    finishHook:_finishHook
343                                       receiveNextMessagesHook:_receiveNextMessagesHook
344                                            responseHeaderHook:_responseHeaderHook
345                                              responseDataHook:_responseDataHook
346                                             responseCloseHook:_responseCloseHook
347                                              didWriteDataHook:_didWriteDataHook];
348  } else {
349    return nil;
350  }
351}
352
353- (void)setStartHook:(void (^)(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
354                               GRPCInterceptorManager *manager))startHook
355              writeDataHook:(void (^)(id data, GRPCInterceptorManager *manager))writeDataHook
356                 finishHook:(void (^)(GRPCInterceptorManager *manager))finishHook
357    receiveNextMessagesHook:(void (^)(NSUInteger numberOfMessages,
358                                      GRPCInterceptorManager *manager))receiveNextMessagesHook
359         responseHeaderHook:(void (^)(NSDictionary *initialMetadata,
360                                      GRPCInterceptorManager *manager))responseHeaderHook
361           responseDataHook:(void (^)(id data, GRPCInterceptorManager *manager))responseDataHook
362          responseCloseHook:(void (^)(NSDictionary *trailingMetadata, NSError *error,
363                                      GRPCInterceptorManager *manager))responseCloseHook
364           didWriteDataHook:(void (^)(GRPCInterceptorManager *manager))didWriteDataHook {
365  _startHook = startHook;
366  _writeDataHook = writeDataHook;
367  _finishHook = finishHook;
368  _receiveNextMessagesHook = receiveNextMessagesHook;
369  _responseHeaderHook = responseHeaderHook;
370  _responseDataHook = responseDataHook;
371  _responseCloseHook = responseCloseHook;
372  _didWriteDataHook = didWriteDataHook;
373}
374
375@end
376
377static GlobalInterceptorFactory *globalInterceptorFactory = nil;
378static dispatch_once_t initGlobalInterceptorFactory;
379
380#pragma mark Tests
381
382@implementation InteropTests {
383  RMTTestService *_service;
384}
385
386#pragma clang diagnostic push
387#pragma clang diagnostic ignored "-Warc-performSelector-leaks"
388- (void)retriableTest:(SEL)selector retries:(int)retries timeout:(NSTimeInterval)timeout {
389  for (int i = 0; i < retries; i++) {
390    NSDate *waitUntil = [[NSDate date] dateByAddingTimeInterval:timeout];
391    NSCondition *cv = [[NSCondition alloc] init];
392    __block BOOL done = NO;
393    [cv lock];
394    dispatch_async(dispatch_get_global_queue(QOS_CLASS_BACKGROUND, 0), ^{
395      [self performSelector:selector];
396      [cv lock];
397      done = YES;
398      [cv signal];
399      [cv unlock];
400    });
401    while (!done && [waitUntil timeIntervalSinceNow] > 0) {
402      [cv waitUntilDate:waitUntil];
403    }
404    if (done) {
405      [cv unlock];
406      break;
407    } else {
408      [cv unlock];
409      [self tearDown];
410      [self setUp];
411    }
412  }
413}
414#pragma clang diagnostic pop
415
416+ (XCTestSuite *)defaultTestSuite {
417  if (self == [InteropTests class]) {
418    return [XCTestSuite testSuiteWithName:@"InteropTestsEmptySuite"];
419  } else {
420    return super.defaultTestSuite;
421  }
422}
423
424+ (NSString *)host {
425  return nil;
426}
427
428// This number indicates how many bytes of overhead does Protocol Buffers encoding add onto the
429// message. The number varies as different message.proto is used on different servers. The actual
430// number for each interop server is overridden in corresponding derived test classes.
431- (int32_t)encodingOverhead {
432  return 0;
433}
434
435// For backwards compatibility
436+ (GRPCTransportType)transportType {
437  return GRPCTransportTypeChttp2BoringSSL;
438}
439
440+ (GRPCTransportID)transport {
441  return NULL;
442}
443
444+ (NSString *)PEMRootCertificates {
445  return nil;
446}
447
448+ (NSString *)hostNameOverride {
449  return nil;
450}
451
452+ (void)setUp {
453  dispatch_once(&initGlobalInterceptorFactory, ^{
454    dispatch_queue_t globalInterceptorQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
455    globalInterceptorFactory =
456        [[GlobalInterceptorFactory alloc] initWithDispatchQueue:globalInterceptorQueue];
457    [GRPCCall2 registerGlobalInterceptor:globalInterceptorFactory];
458  });
459}
460
461- (void)setUp {
462  self.continueAfterFailure = NO;
463
464  [GRPCCall resetHostSettings];
465
466#pragma clang diagnostic push
467#pragma clang diagnostic ignored "-Wdeprecated-declarations"
468  [GRPCCall closeOpenConnections];
469#pragma clang diagnostic pop
470
471  _service = [[self class] host] ? [RMTTestService serviceWithHost:[[self class] host]] : nil;
472}
473
474- (void)testEmptyUnaryRPC {
475  XCTAssertNotNil([[self class] host]);
476  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"EmptyUnary"];
477
478  GPBEmpty *request = [GPBEmpty message];
479
480  [_service emptyCallWithRequest:request
481                         handler:^(GPBEmpty *response, NSError *error) {
482                           XCTAssertNil(error, @"Finished with unexpected error: %@", error);
483
484                           id expectedResponse = [GPBEmpty message];
485                           XCTAssertEqualObjects(response, expectedResponse);
486
487                           [expectation fulfill];
488                         }];
489
490  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
491}
492
493- (void)testEmptyUnaryRPCWithV2API {
494  XCTAssertNotNil([[self class] host]);
495  __weak XCTestExpectation *expectReceive =
496      [self expectationWithDescription:@"EmptyUnaryWithV2API received message"];
497  __weak XCTestExpectation *expectComplete =
498      [self expectationWithDescription:@"EmptyUnaryWithV2API completed"];
499
500  GPBEmpty *request = [GPBEmpty message];
501  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
502  // For backwards compatibility
503  options.transportType = [[self class] transportType];
504  options.transport = [[self class] transport];
505  options.PEMRootCertificates = [[self class] PEMRootCertificates];
506  options.hostNameOverride = [[self class] hostNameOverride];
507
508  GRPCUnaryProtoCall *call = [_service
509      emptyCallWithMessage:request
510           responseHandler:[[InteropTestsBlockCallbacks alloc] initWithInitialMetadataCallback:nil
511                               messageCallback:^(id message) {
512                                 if (message) {
513                                   id expectedResponse = [GPBEmpty message];
514                                   XCTAssertEqualObjects(message, expectedResponse);
515                                   [expectReceive fulfill];
516                                 }
517                               }
518                               closeCallback:^(NSDictionary *trailingMetadata, NSError *error) {
519                                 XCTAssertNil(error, @"Unexpected error: %@", error);
520                                 [expectComplete fulfill];
521                               }]
522               callOptions:options];
523  [call start];
524  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
525}
526
527// Test that responses can be dispatched even if we do not run main run-loop
528- (void)testAsyncDispatchWithV2API {
529  XCTAssertNotNil([[self class] host]);
530
531  GPBEmpty *request = [GPBEmpty message];
532  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
533  // For backwards compatibility
534  options.transportType = [[self class] transportType];
535  options.transport = [[self class] transport];
536  options.PEMRootCertificates = [[self class] PEMRootCertificates];
537  options.hostNameOverride = [[self class] hostNameOverride];
538
539  __block BOOL messageReceived = NO;
540  __block BOOL done = NO;
541  NSCondition *cond = [[NSCondition alloc] init];
542  GRPCUnaryProtoCall *call = [_service
543      emptyCallWithMessage:request
544           responseHandler:[[InteropTestsBlockCallbacks alloc] initWithInitialMetadataCallback:nil
545                               messageCallback:^(id message) {
546                                 if (message) {
547                                   id expectedResponse = [GPBEmpty message];
548                                   XCTAssertEqualObjects(message, expectedResponse);
549                                   [cond lock];
550                                   messageReceived = YES;
551                                   [cond unlock];
552                                 }
553                               }
554                               closeCallback:^(NSDictionary *trailingMetadata, NSError *error) {
555                                 XCTAssertNil(error, @"Unexpected error: %@", error);
556                                 [cond lock];
557                                 done = YES;
558                                 [cond signal];
559                                 [cond unlock];
560                               }]
561               callOptions:options];
562
563  NSDate *deadline = [NSDate dateWithTimeIntervalSinceNow:TEST_TIMEOUT];
564  [call start];
565
566  [cond lock];
567  while (!done && [deadline timeIntervalSinceNow] > 0) {
568    [cond waitUntilDate:deadline];
569  }
570  XCTAssertTrue(messageReceived);
571  XCTAssertTrue(done);
572  [cond unlock];
573}
574
575- (void)testLargeUnaryRPC {
576  XCTAssertNotNil([[self class] host]);
577  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"LargeUnary"];
578
579  RMTSimpleRequest *request = [RMTSimpleRequest message];
580  request.responseType = RMTPayloadType_Compressable;
581  request.responseSize = 314159;
582  request.payload.body = [NSMutableData dataWithLength:271828];
583
584  [_service unaryCallWithRequest:request
585                         handler:^(RMTSimpleResponse *response, NSError *error) {
586                           XCTAssertNil(error, @"Finished with unexpected error: %@", error);
587
588                           RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
589                           expectedResponse.payload.type = RMTPayloadType_Compressable;
590                           expectedResponse.payload.body = [NSMutableData dataWithLength:314159];
591                           XCTAssertEqualObjects(response, expectedResponse);
592
593                           [expectation fulfill];
594                         }];
595
596  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
597}
598
599- (void)testUnaryResponseHandler {
600  XCTAssertNotNil([[self class] host]);
601  // The test does not work on a remote server since it does not echo a trailer
602  if ([[self class] isRemoteTest]) return;
603  XCTestExpectation *expectComplete = [self expectationWithDescription:@"call complete"];
604  XCTestExpectation *expectCompleteMainQueue =
605      [self expectationWithDescription:@"main queue call complete"];
606
607  RMTSimpleRequest *request = [RMTSimpleRequest message];
608  request.responseType = RMTPayloadType_Compressable;
609  request.responseSize = 314159;
610  request.payload.body = [NSMutableData dataWithLength:271828];
611
612  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
613  // For backwards compatibility
614  options.transportType = [[self class] transportType];
615  options.transport = [[self class] transport];
616  options.PEMRootCertificates = [[self class] PEMRootCertificates];
617  options.hostNameOverride = [[self class] hostNameOverride];
618  const unsigned char raw_bytes[] = {1, 2, 3, 4};
619  NSData *trailer_data = [NSData dataWithBytes:raw_bytes length:sizeof(raw_bytes)];
620  options.initialMetadata = @{
621    @"x-grpc-test-echo-trailing-bin" : trailer_data,
622    @"x-grpc-test-echo-initial" : @"test-header"
623  };
624
625  __block GRPCUnaryResponseHandler *handler = [[GRPCUnaryResponseHandler alloc]
626      initWithResponseHandler:^(GPBMessage *response, NSError *error) {
627        XCTAssertNil(error, @"Unexpected error: %@", error);
628        RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
629        expectedResponse.payload.type = RMTPayloadType_Compressable;
630        expectedResponse.payload.body = [NSMutableData dataWithLength:314159];
631        XCTAssertEqualObjects(response, expectedResponse);
632        XCTAssertEqualObjects(handler.responseHeaders[@"x-grpc-test-echo-initial"], @"test-header");
633        XCTAssertEqualObjects(handler.responseTrailers[@"x-grpc-test-echo-trailing-bin"],
634                              trailer_data);
635        [expectComplete fulfill];
636      }
637        responseDispatchQueue:dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL)];
638  __block GRPCUnaryResponseHandler *handlerMainQueue = [[GRPCUnaryResponseHandler alloc]
639      initWithResponseHandler:^(GPBMessage *response, NSError *error) {
640        XCTAssertNil(error, @"Unexpected error: %@", error);
641        RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
642        expectedResponse.payload.type = RMTPayloadType_Compressable;
643        expectedResponse.payload.body = [NSMutableData dataWithLength:314159];
644        XCTAssertEqualObjects(response, expectedResponse);
645        XCTAssertEqualObjects(handlerMainQueue.responseHeaders[@"x-grpc-test-echo-initial"],
646                              @"test-header");
647        XCTAssertEqualObjects(handlerMainQueue.responseTrailers[@"x-grpc-test-echo-trailing-bin"],
648                              trailer_data);
649        [expectCompleteMainQueue fulfill];
650      }
651        responseDispatchQueue:nil];
652
653  [[_service unaryCallWithMessage:request responseHandler:handler callOptions:options] start];
654  [[_service unaryCallWithMessage:request responseHandler:handlerMainQueue
655                      callOptions:options] start];
656  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
657}
658
659- (void)testLargeUnaryRPCWithV2API {
660  XCTAssertNotNil([[self class] host]);
661  __weak XCTestExpectation *expectReceive =
662      [self expectationWithDescription:@"LargeUnaryWithV2API received message"];
663  __weak XCTestExpectation *expectComplete =
664      [self expectationWithDescription:@"LargeUnaryWithV2API received complete"];
665
666  RMTSimpleRequest *request = [RMTSimpleRequest message];
667  request.responseType = RMTPayloadType_Compressable;
668  request.responseSize = 314159;
669  request.payload.body = [NSMutableData dataWithLength:271828];
670
671  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
672  // For backwards compatibility
673  options.transportType = [[self class] transportType];
674  options.transport = [[self class] transport];
675  options.PEMRootCertificates = [[self class] PEMRootCertificates];
676  options.hostNameOverride = [[self class] hostNameOverride];
677
678  GRPCUnaryProtoCall *call = [_service
679      unaryCallWithMessage:request
680           responseHandler:[[InteropTestsBlockCallbacks alloc] initWithInitialMetadataCallback:nil
681                               messageCallback:^(id message) {
682                                 XCTAssertNotNil(message);
683                                 if (message) {
684                                   RMTSimpleResponse *expectedResponse =
685                                       [RMTSimpleResponse message];
686                                   expectedResponse.payload.type = RMTPayloadType_Compressable;
687                                   expectedResponse.payload.body =
688                                       [NSMutableData dataWithLength:314159];
689                                   XCTAssertEqualObjects(message, expectedResponse);
690
691                                   [expectReceive fulfill];
692                                 }
693                               }
694                               closeCallback:^(NSDictionary *trailingMetadata, NSError *error) {
695                                 XCTAssertNil(error, @"Unexpected error: %@", error);
696                                 [expectComplete fulfill];
697                               }]
698               callOptions:options];
699  [call start];
700  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
701}
702
703- (void)testConcurrentRPCsWithErrorsWithV2API {
704  NSMutableArray *completeExpectations = [NSMutableArray array];
705  NSMutableArray *calls = [NSMutableArray array];
706  int num_rpcs = 10;
707  for (int i = 0; i < num_rpcs; ++i) {
708    [completeExpectations
709        addObject:[self expectationWithDescription:
710                            [NSString stringWithFormat:@"Received trailer for RPC %d", i]]];
711
712    RMTSimpleRequest *request = [RMTSimpleRequest message];
713    request.responseType = RMTPayloadType_Compressable;
714    request.responseSize = 314159;
715    request.payload.body = [NSMutableData dataWithLength:271828];
716    if (i % 3 == 0) {
717      request.responseStatus.code = GRPC_STATUS_UNAVAILABLE;
718    } else if (i % 7 == 0) {
719      request.responseStatus.code = GRPC_STATUS_CANCELLED;
720    }
721    GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
722    // For backwards compatibility
723    options.transportType = [[self class] transportType];
724    options.transport = [[self class] transport];
725    options.PEMRootCertificates = [[self class] PEMRootCertificates];
726    options.hostNameOverride = [[self class] hostNameOverride];
727
728    GRPCUnaryProtoCall *call = [_service
729        unaryCallWithMessage:request
730             responseHandler:[[InteropTestsBlockCallbacks alloc] initWithInitialMetadataCallback:nil
731                                 messageCallback:^(id message) {
732                                   if (message) {
733                                     RMTSimpleResponse *expectedResponse =
734                                         [RMTSimpleResponse message];
735                                     expectedResponse.payload.type = RMTPayloadType_Compressable;
736                                     expectedResponse.payload.body =
737                                         [NSMutableData dataWithLength:314159];
738                                     XCTAssertEqualObjects(message, expectedResponse);
739                                   }
740                                 }
741                                 closeCallback:^(NSDictionary *trailingMetadata, NSError *error) {
742                                   [completeExpectations[i] fulfill];
743                                 }]
744                 callOptions:options];
745    [calls addObject:call];
746  }
747
748  for (int i = 0; i < num_rpcs; ++i) {
749    GRPCUnaryProtoCall *call = calls[i];
750    [call start];
751  }
752  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
753}
754
755- (void)concurrentRPCsWithErrors {
756  const int kNumRpcs = 10;
757  __block int completedCallCount = 0;
758  NSCondition *cv = [[NSCondition alloc] init];
759  NSDate *waitUntil = [[NSDate date] dateByAddingTimeInterval:TEST_TIMEOUT];
760  [cv lock];
761  for (int i = 0; i < kNumRpcs; ++i) {
762    RMTSimpleRequest *request = [RMTSimpleRequest message];
763    request.responseType = RMTPayloadType_Compressable;
764    request.responseSize = 314159;
765    request.payload.body = [NSMutableData dataWithLength:271828];
766    if (i % 3 == 0) {
767      request.responseStatus.code = GRPC_STATUS_UNAVAILABLE;
768    } else if (i % 7 == 0) {
769      request.responseStatus.code = GRPC_STATUS_CANCELLED;
770    }
771
772    GRPCProtoCall *call = [_service
773        RPCToUnaryCallWithRequest:request
774                          handler:^(RMTSimpleResponse *response, NSError *error) {
775                            if (error == nil) {
776                              RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
777                              expectedResponse.payload.type = RMTPayloadType_Compressable;
778                              expectedResponse.payload.body = [NSMutableData dataWithLength:314159];
779                              XCTAssertEqualObjects(response, expectedResponse);
780                            }
781                            // DEBUG
782                            [cv lock];
783                            if (++completedCallCount == kNumRpcs) {
784                              [cv signal];
785                            }
786                            [cv unlock];
787                          }];
788    [call setResponseDispatchQueue:dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL)];
789    [call start];
790  }
791  while (completedCallCount<kNumRpcs && [waitUntil timeIntervalSinceNow]> 0) {
792    [cv waitUntilDate:waitUntil];
793  }
794  [cv unlock];
795}
796
797- (void)testConcurrentRPCsWithErrors {
798  [self retriableTest:@selector(concurrentRPCsWithErrors) retries:kTestRetries timeout:10];
799}
800
801- (void)testPacketCoalescing {
802  XCTAssertNotNil([[self class] host]);
803  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"LargeUnary"];
804
805  RMTSimpleRequest *request = [RMTSimpleRequest message];
806  request.responseType = RMTPayloadType_Compressable;
807  request.responseSize = 10;
808  request.payload.body = [NSMutableData dataWithLength:10];
809
810  [GRPCCall enableOpBatchLog:YES];
811  [_service unaryCallWithRequest:request
812                         handler:^(RMTSimpleResponse *response, NSError *error) {
813                           XCTAssertNil(error, @"Finished with unexpected error: %@", error);
814
815                           RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
816                           expectedResponse.payload.type = RMTPayloadType_Compressable;
817                           expectedResponse.payload.body = [NSMutableData dataWithLength:10];
818                           XCTAssertEqualObjects(response, expectedResponse);
819
820                           // The test is a success if there is a batch of exactly 3 ops
821                           // (SEND_INITIAL_METADATA, SEND_MESSAGE, SEND_CLOSE_FROM_CLIENT). Without
822                           // packet coalescing each batch of ops contains only one op.
823                           NSArray *opBatches = [GRPCCall obtainAndCleanOpBatchLog];
824                           const NSInteger kExpectedOpBatchSize = 3;
825                           for (NSObject *o in opBatches) {
826                             if ([o isKindOfClass:[NSArray class]]) {
827                               NSArray *batch = (NSArray *)o;
828                               if ([batch count] == kExpectedOpBatchSize) {
829                                 [expectation fulfill];
830                                 break;
831                               }
832                             }
833                           }
834                         }];
835
836  [self waitForExpectationsWithTimeout:16 handler:nil];
837  [GRPCCall enableOpBatchLog:NO];
838}
839
840- (void)test4MBResponsesAreAccepted {
841  XCTAssertNotNil([[self class] host]);
842  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"MaxResponseSize"];
843
844  RMTSimpleRequest *request = [RMTSimpleRequest message];
845  const int32_t kPayloadSize = 4 * 1024 * 1024 - self.encodingOverhead;  // 4MB - encoding overhead
846  request.responseSize = kPayloadSize;
847
848  [_service unaryCallWithRequest:request
849                         handler:^(RMTSimpleResponse *response, NSError *error) {
850                           XCTAssertNil(error, @"Finished with unexpected error: %@", error);
851                           XCTAssertEqual(response.payload.body.length, kPayloadSize);
852                           [expectation fulfill];
853                         }];
854
855  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
856}
857
858- (void)testResponsesOverMaxSizeFailWithActionableMessage {
859  XCTAssertNotNil([[self class] host]);
860  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"ResponseOverMaxSize"];
861
862  RMTSimpleRequest *request = [RMTSimpleRequest message];
863  const int32_t kPayloadSize = 4 * 1024 * 1024 - self.encodingOverhead + 1;  // 1B over max size
864  request.responseSize = kPayloadSize;
865
866  [_service unaryCallWithRequest:request
867                         handler:^(RMTSimpleResponse *response, NSError *error) {
868                           // TODO(jcanizales): Catch the error and rethrow it with an actionable
869                           // message:
870                           // - Use +[GRPCCall setResponseSizeLimit:forHost:] to set a higher limit.
871                           // - If you're developing the server, consider using response streaming,
872                           // or let clients filter
873                           //   responses by setting a google.protobuf.FieldMask in the request:
874                           //   https://github.com/google/protobuf/blob/master/src/google/protobuf/field_mask.proto
875                           XCTAssertEqualObjects(
876                               error.localizedDescription,
877                               @"Received message larger than max (4194305 vs. 4194304)");
878                           [expectation fulfill];
879                         }];
880
881  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
882}
883
884- (void)testResponsesOver4MBAreAcceptedIfOptedIn {
885  XCTAssertNotNil([[self class] host]);
886  __weak XCTestExpectation *expectation =
887      [self expectationWithDescription:@"HigherResponseSizeLimit"];
888  __block NSError *callError = nil;
889
890  RMTSimpleRequest *request = [RMTSimpleRequest message];
891  const size_t kPayloadSize = 5 * 1024 * 1024;  // 5MB
892  request.responseSize = kPayloadSize;
893
894  [GRPCCall setResponseSizeLimit:6 * 1024 * 1024 forHost:[[self class] host]];
895  [_service unaryCallWithRequest:request
896                         handler:^(RMTSimpleResponse *response, NSError *error) {
897                           callError = error;
898                           [expectation fulfill];
899                         }];
900
901  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
902  XCTAssertNil(callError, @"Finished with unexpected error: %@", callError);
903}
904
905- (void)testClientStreamingRPC {
906  XCTAssertNotNil([[self class] host]);
907  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"ClientStreaming"];
908
909  RMTStreamingInputCallRequest *request1 = [RMTStreamingInputCallRequest message];
910  request1.payload.body = [NSMutableData dataWithLength:27182];
911
912  RMTStreamingInputCallRequest *request2 = [RMTStreamingInputCallRequest message];
913  request2.payload.body = [NSMutableData dataWithLength:8];
914
915  RMTStreamingInputCallRequest *request3 = [RMTStreamingInputCallRequest message];
916  request3.payload.body = [NSMutableData dataWithLength:1828];
917
918  RMTStreamingInputCallRequest *request4 = [RMTStreamingInputCallRequest message];
919  request4.payload.body = [NSMutableData dataWithLength:45904];
920
921  GRXWriter *writer = [GRXWriter writerWithContainer:@[ request1, request2, request3, request4 ]];
922
923  [_service streamingInputCallWithRequestsWriter:writer
924                                         handler:^(RMTStreamingInputCallResponse *response,
925                                                   NSError *error) {
926                                           XCTAssertNil(
927                                               error, @"Finished with unexpected error: %@", error);
928
929                                           RMTStreamingInputCallResponse *expectedResponse =
930                                               [RMTStreamingInputCallResponse message];
931                                           expectedResponse.aggregatedPayloadSize = 74922;
932                                           XCTAssertEqualObjects(response, expectedResponse);
933
934                                           [expectation fulfill];
935                                         }];
936
937  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
938}
939
940- (void)testServerStreamingRPC {
941  XCTAssertNotNil([[self class] host]);
942  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"ServerStreaming"];
943
944  NSArray *expectedSizes = @[ @31415, @9, @2653, @58979 ];
945
946  RMTStreamingOutputCallRequest *request = [RMTStreamingOutputCallRequest message];
947  for (NSNumber *size in expectedSizes) {
948    RMTResponseParameters *parameters = [RMTResponseParameters message];
949    parameters.size = [size intValue];
950    [request.responseParametersArray addObject:parameters];
951  }
952
953  __block int index = 0;
954  [_service
955      streamingOutputCallWithRequest:request
956                        eventHandler:^(BOOL done, RMTStreamingOutputCallResponse *response,
957                                       NSError *error) {
958                          XCTAssertNil(error, @"Finished with unexpected error: %@", error);
959                          XCTAssertTrue(done || response,
960                                        @"Event handler called without an event.");
961
962                          if (response) {
963                            XCTAssertLessThan(index, 4, @"More than 4 responses received.");
964                            id expected = [RMTStreamingOutputCallResponse
965                                messageWithPayloadSize:expectedSizes[index]];
966                            XCTAssertEqualObjects(response, expected);
967                            index += 1;
968                          }
969
970                          if (done) {
971                            XCTAssertEqual(index, 4, @"Received %i responses instead of 4.", index);
972                            [expectation fulfill];
973                          }
974                        }];
975
976  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
977}
978
979- (void)testPingPongRPC {
980  XCTAssertNotNil([[self class] host]);
981  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"PingPong"];
982
983  NSArray *requests = @[ @27182, @8, @1828, @45904 ];
984  NSArray *responses = @[ @31415, @9, @2653, @58979 ];
985
986  GRXBufferedPipe *requestsBuffer = [[GRXBufferedPipe alloc] init];
987
988  __block int index = 0;
989
990  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
991                                               requestedResponseSize:responses[index]];
992  [requestsBuffer writeValue:request];
993
994  [_service fullDuplexCallWithRequestsWriter:requestsBuffer
995                                eventHandler:^(BOOL done, RMTStreamingOutputCallResponse *response,
996                                               NSError *error) {
997                                  XCTAssertNil(error, @"Finished with unexpected error: %@", error);
998                                  XCTAssertTrue(done || response,
999                                                @"Event handler called without an event.");
1000
1001                                  if (response) {
1002                                    XCTAssertLessThan(index, 4, @"More than 4 responses received.");
1003                                    id expected = [RMTStreamingOutputCallResponse
1004                                        messageWithPayloadSize:responses[index]];
1005                                    XCTAssertEqualObjects(response, expected);
1006                                    index += 1;
1007                                    if (index < 4) {
1008                                      id request = [RMTStreamingOutputCallRequest
1009                                          messageWithPayloadSize:requests[index]
1010                                           requestedResponseSize:responses[index]];
1011                                      [requestsBuffer writeValue:request];
1012                                    } else {
1013                                      [requestsBuffer writesFinishedWithError:nil];
1014                                    }
1015                                  }
1016
1017                                  if (done) {
1018                                    XCTAssertEqual(index, 4, @"Received %i responses instead of 4.",
1019                                                   index);
1020                                    [expectation fulfill];
1021                                  }
1022                                }];
1023  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1024}
1025
1026- (void)testPingPongRPCWithV2API {
1027  XCTAssertNotNil([[self class] host]);
1028  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"PingPongWithV2API"];
1029
1030  NSArray *requests = @[ @27182, @8, @1828, @45904 ];
1031  NSArray *responses = @[ @31415, @9, @2653, @58979 ];
1032
1033  __block int index = 0;
1034
1035  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1036                                               requestedResponseSize:responses[index]];
1037  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1038  // For backwards compatibility
1039  options.transportType = [[self class] transportType];
1040  options.transport = [[self class] transport];
1041  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1042  options.hostNameOverride = [[self class] hostNameOverride];
1043
1044  __block GRPCStreamingProtoCall *call = [_service
1045      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1046                                            initWithInitialMetadataCallback:nil
1047                                            messageCallback:^(id message) {
1048                                              XCTAssertLessThan(index, 4,
1049                                                                @"More than 4 responses received.");
1050                                              id expected = [RMTStreamingOutputCallResponse
1051                                                  messageWithPayloadSize:responses[index]];
1052                                              XCTAssertEqualObjects(message, expected);
1053                                              index += 1;
1054                                              if (index < 4) {
1055                                                id request = [RMTStreamingOutputCallRequest
1056                                                    messageWithPayloadSize:requests[index]
1057                                                     requestedResponseSize:responses[index]];
1058                                                [call writeMessage:request];
1059                                              } else {
1060                                                [call finish];
1061                                              }
1062                                            }
1063                                            closeCallback:^(NSDictionary *trailingMetadata,
1064                                                            NSError *error) {
1065                                              XCTAssertNil(error,
1066                                                           @"Finished with unexpected error: %@",
1067                                                           error);
1068                                              XCTAssertEqual(index, 4,
1069                                                             @"Received %i responses instead of 4.",
1070                                                             index);
1071                                              [expectation fulfill];
1072                                            }]
1073                            callOptions:options];
1074  [call start];
1075  [call writeMessage:request];
1076
1077  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1078}
1079
1080- (void)testPingPongRPCWithFlowControl {
1081  XCTAssertNotNil([[self class] host]);
1082  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"PingPongWithV2API"];
1083
1084  NSArray *requests = @[ @27182, @8, @1828, @45904 ];
1085  NSArray *responses = @[ @31415, @9, @2653, @58979 ];
1086
1087  __block int index = 0;
1088
1089  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1090                                               requestedResponseSize:responses[index]];
1091  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1092  // For backwards compatibility
1093  options.transportType = [[self class] transportType];
1094  options.transport = [[self class] transport];
1095  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1096  options.hostNameOverride = [[self class] hostNameOverride];
1097  options.flowControlEnabled = YES;
1098  __block BOOL canWriteData = NO;
1099
1100  __block GRPCStreamingProtoCall *call = [_service
1101      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1102                                            initWithInitialMetadataCallback:nil
1103                                            messageCallback:^(id message) {
1104                                              XCTAssertLessThan(index, 4,
1105                                                                @"More than 4 responses received.");
1106                                              id expected = [RMTStreamingOutputCallResponse
1107                                                  messageWithPayloadSize:responses[index]];
1108                                              XCTAssertEqualObjects(message, expected);
1109                                              index += 1;
1110                                              if (index < 4) {
1111                                                id request = [RMTStreamingOutputCallRequest
1112                                                    messageWithPayloadSize:requests[index]
1113                                                     requestedResponseSize:responses[index]];
1114                                                XCTAssertTrue(canWriteData);
1115                                                canWriteData = NO;
1116                                                [call writeMessage:request];
1117                                                [call receiveNextMessage];
1118                                              } else {
1119                                                [call finish];
1120                                              }
1121                                            }
1122                                            closeCallback:^(NSDictionary *trailingMetadata,
1123                                                            NSError *error) {
1124                                              XCTAssertNil(error,
1125                                                           @"Finished with unexpected error: %@",
1126                                                           error);
1127                                              XCTAssertEqual(index, 4,
1128                                                             @"Received %i responses instead of 4.",
1129                                                             index);
1130                                              [expectation fulfill];
1131                                            }
1132                                            writeMessageCallback:^{
1133                                              canWriteData = YES;
1134                                            }]
1135                            callOptions:options];
1136  [call start];
1137  [call receiveNextMessage];
1138  [call writeMessage:request];
1139
1140  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1141}
1142
1143- (void)testEmptyStreamRPC {
1144  XCTAssertNotNil([[self class] host]);
1145  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"EmptyStream"];
1146  [_service fullDuplexCallWithRequestsWriter:[GRXWriter emptyWriter]
1147                                eventHandler:^(BOOL done, RMTStreamingOutputCallResponse *response,
1148                                               NSError *error) {
1149                                  XCTAssertNil(error, @"Finished with unexpected error: %@", error);
1150                                  XCTAssert(done, @"Unexpected response: %@", response);
1151                                  [expectation fulfill];
1152                                }];
1153  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1154}
1155
1156- (void)testCancelAfterBeginRPC {
1157  XCTAssertNotNil([[self class] host]);
1158  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"CancelAfterBegin"];
1159
1160  // A buffered pipe to which we never write any value acts as a writer that just hangs.
1161  GRXBufferedPipe *requestsBuffer = [[GRXBufferedPipe alloc] init];
1162
1163  GRPCProtoCall *call = [_service
1164      RPCToStreamingInputCallWithRequestsWriter:requestsBuffer
1165                                        handler:^(RMTStreamingInputCallResponse *response,
1166                                                  NSError *error) {
1167                                          XCTAssertEqual(error.code, GRPC_STATUS_CANCELLED);
1168                                          [expectation fulfill];
1169                                        }];
1170  XCTAssertEqual(call.state, GRXWriterStateNotStarted);
1171
1172  [call start];
1173  XCTAssertEqual(call.state, GRXWriterStateStarted);
1174
1175  [call cancel];
1176  XCTAssertEqual(call.state, GRXWriterStateFinished);
1177
1178  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1179}
1180
1181- (void)testCancelAfterBeginRPCWithV2API {
1182  XCTAssertNotNil([[self class] host]);
1183  __weak XCTestExpectation *expectation =
1184      [self expectationWithDescription:@"CancelAfterBeginWithV2API"];
1185
1186  // A buffered pipe to which we never write any value acts as a writer that just hangs.
1187  __block GRPCStreamingProtoCall *call = [_service
1188      streamingInputCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1189                                                initWithInitialMetadataCallback:nil
1190                                                messageCallback:^(id message) {
1191                                                  XCTFail(@"Not expected to receive message");
1192                                                }
1193                                                closeCallback:^(NSDictionary *trailingMetadata,
1194                                                                NSError *error) {
1195                                                  XCTAssertEqual(error.code, GRPC_STATUS_CANCELLED);
1196                                                  [expectation fulfill];
1197                                                }]
1198                                callOptions:nil];
1199  [call start];
1200  [call cancel];
1201
1202  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1203}
1204
1205- (void)testCancelAfterFirstResponseRPC {
1206  XCTAssertNotNil([[self class] host]);
1207  __weak XCTestExpectation *expectation =
1208      [self expectationWithDescription:@"CancelAfterFirstResponse"];
1209
1210  // A buffered pipe to which we write a single value but never close
1211  GRXBufferedPipe *requestsBuffer = [[GRXBufferedPipe alloc] init];
1212
1213  __block BOOL receivedResponse = NO;
1214
1215  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:@21782
1216                                               requestedResponseSize:@31415];
1217
1218  [requestsBuffer writeValue:request];
1219
1220  __block GRPCProtoCall *call = [_service
1221      RPCToFullDuplexCallWithRequestsWriter:requestsBuffer
1222                               eventHandler:^(BOOL done, RMTStreamingOutputCallResponse *response,
1223                                              NSError *error) {
1224                                 if (receivedResponse) {
1225                                   XCTAssert(done, @"Unexpected extra response %@", response);
1226                                   XCTAssertEqual(error.code, GRPC_STATUS_CANCELLED);
1227                                   [expectation fulfill];
1228                                 } else {
1229                                   XCTAssertNil(error, @"Finished with unexpected error: %@",
1230                                                error);
1231                                   XCTAssertFalse(done, @"Finished without response");
1232                                   XCTAssertNotNil(response);
1233                                   receivedResponse = YES;
1234                                   [call cancel];
1235                                 }
1236                               }];
1237  [call start];
1238  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1239}
1240
1241- (void)testCancelAfterFirstResponseRPCWithV2API {
1242  XCTAssertNotNil([[self class] host]);
1243  __weak XCTestExpectation *completionExpectation =
1244      [self expectationWithDescription:@"Call completed."];
1245  __weak XCTestExpectation *responseExpectation =
1246      [self expectationWithDescription:@"Received response."];
1247
1248  __block BOOL receivedResponse = NO;
1249
1250  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1251  // For backwards compatibility
1252  options.transportType = self.class.transportType;
1253  options.transport = [[self class] transport];
1254  options.PEMRootCertificates = self.class.PEMRootCertificates;
1255  options.hostNameOverride = [[self class] hostNameOverride];
1256
1257  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:@21782
1258                                               requestedResponseSize:@31415];
1259
1260  __block GRPCStreamingProtoCall *call = [_service
1261      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1262                                            initWithInitialMetadataCallback:nil
1263                                            messageCallback:^(id message) {
1264                                              XCTAssertFalse(receivedResponse);
1265                                              receivedResponse = YES;
1266                                              [call cancel];
1267                                              [responseExpectation fulfill];
1268                                            }
1269                                            closeCallback:^(NSDictionary *trailingMetadata,
1270                                                            NSError *error) {
1271                                              XCTAssertEqual(error.code, GRPC_STATUS_CANCELLED);
1272                                              [completionExpectation fulfill];
1273                                            }]
1274                            callOptions:options];
1275  [call start];
1276  [call writeMessage:request];
1277  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1278}
1279
1280- (void)testCancelAfterFirstRequestWithV2API {
1281  XCTAssertNotNil([[self class] host]);
1282  __weak XCTestExpectation *completionExpectation =
1283      [self expectationWithDescription:@"Call completed."];
1284
1285  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1286  // For backwards compatibility
1287  options.transportType = self.class.transportType;
1288  options.transport = [[self class] transport];
1289  options.PEMRootCertificates = self.class.PEMRootCertificates;
1290  options.hostNameOverride = [[self class] hostNameOverride];
1291
1292  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:@21782
1293                                               requestedResponseSize:@31415];
1294
1295  __block GRPCStreamingProtoCall *call = [_service
1296      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1297                                            initWithInitialMetadataCallback:nil
1298                                            messageCallback:^(id message) {
1299                                              XCTFail(@"Received unexpected response.");
1300                                            }
1301                                            closeCallback:^(NSDictionary *trailingMetadata,
1302                                                            NSError *error) {
1303                                              XCTAssertEqual(error.code, GRPC_STATUS_CANCELLED);
1304                                              [completionExpectation fulfill];
1305                                            }]
1306                            callOptions:options];
1307  [call start];
1308  [call writeMessage:request];
1309  [call cancel];
1310  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1311}
1312
1313- (void)testRPCAfterClosingOpenConnections {
1314  XCTAssertNotNil([[self class] host]);
1315  __weak XCTestExpectation *expectation =
1316      [self expectationWithDescription:@"RPC after closing connection"];
1317
1318  GPBEmpty *request = [GPBEmpty message];
1319
1320  [_service
1321      emptyCallWithRequest:request
1322                   handler:^(GPBEmpty *response, NSError *error) {
1323                     XCTAssertNil(error, @"First RPC finished with unexpected error: %@", error);
1324
1325#pragma clang diagnostic push
1326#pragma clang diagnostic ignored "-Wdeprecated-declarations"
1327                     [GRPCCall closeOpenConnections];
1328#pragma clang diagnostic pop
1329
1330                     [self->_service
1331                         emptyCallWithRequest:request
1332                                      handler:^(GPBEmpty *response, NSError *error) {
1333                                        XCTAssertNil(
1334                                            error, @"Second RPC finished with unexpected error: %@",
1335                                            error);
1336                                        [expectation fulfill];
1337                                      }];
1338                   }];
1339
1340  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1341}
1342
1343- (void)testCompressedUnaryRPC {
1344  // This test needs to be disabled for remote test because interop server grpc-test
1345  // does not support compression.
1346  if (isRemoteInteropTest([[self class] host])) {
1347    return;
1348  }
1349  XCTAssertNotNil([[self class] host]);
1350  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"LargeUnary"];
1351
1352  RMTSimpleRequest *request = [RMTSimpleRequest message];
1353  request.responseType = RMTPayloadType_Compressable;
1354  request.responseSize = 314159;
1355  request.payload.body = [NSMutableData dataWithLength:271828];
1356  request.expectCompressed.value = YES;
1357  [GRPCCall setDefaultCompressMethod:GRPCCompressGzip forhost:[[self class] host]];
1358
1359  [_service unaryCallWithRequest:request
1360                         handler:^(RMTSimpleResponse *response, NSError *error) {
1361                           XCTAssertNil(error, @"Finished with unexpected error: %@", error);
1362
1363                           RMTSimpleResponse *expectedResponse = [RMTSimpleResponse message];
1364                           expectedResponse.payload.type = RMTPayloadType_Compressable;
1365                           expectedResponse.payload.body = [NSMutableData dataWithLength:314159];
1366                           XCTAssertEqualObjects(response, expectedResponse);
1367
1368                           [expectation fulfill];
1369                         }];
1370
1371  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1372}
1373
1374- (void)testKeepaliveWithV2API {
1375  XCTAssertNotNil([[self class] host]);
1376  if ([[self class] transport] == gGRPCCoreCronetID) {
1377    // Cronet does not support keepalive
1378    return;
1379  }
1380  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"Keepalive"];
1381
1382  NSNumber *kRequestSize = @27182;
1383  NSNumber *kResponseSize = @31415;
1384
1385  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:kRequestSize
1386                                               requestedResponseSize:kResponseSize];
1387  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1388  options.transportType = [[self class] transportType];
1389  options.transport = [[self class] transport];
1390  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1391  options.hostNameOverride = [[self class] hostNameOverride];
1392  options.keepaliveInterval = 1.5;
1393  options.keepaliveTimeout = 0;
1394
1395  __block GRPCStreamingProtoCall *call = [_service
1396      fullDuplexCallWithResponseHandler:
1397          [[InteropTestsBlockCallbacks alloc]
1398              initWithInitialMetadataCallback:nil
1399                              messageCallback:nil
1400                                closeCallback:^(NSDictionary *trailingMetadata, NSError *error) {
1401                                  XCTAssertNotNil(error);
1402                                  XCTAssertEqual(
1403                                      error.code, GRPC_STATUS_UNAVAILABLE,
1404                                      @"Received status %ld instead of UNAVAILABLE (14).",
1405                                      error.code);
1406                                  [expectation fulfill];
1407                                }]
1408                            callOptions:options];
1409  [call writeMessage:request];
1410  [call start];
1411
1412  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1413  [call finish];
1414}
1415
1416- (void)testDefaultInterceptor {
1417  XCTAssertNotNil([[self class] host]);
1418  __weak XCTestExpectation *expectation =
1419      [self expectationWithDescription:@"testDefaultInterceptor"];
1420
1421  NSArray *requests = @[ @27182, @8, @1828, @45904 ];
1422  NSArray *responses = @[ @31415, @9, @2653, @58979 ];
1423
1424  __block int index = 0;
1425
1426  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1427                                               requestedResponseSize:responses[index]];
1428  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1429  // For backwards compatibility
1430  options.transportType = [[self class] transportType];
1431  options.transport = [[self class] transport];
1432  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1433  options.hostNameOverride = [[self class] hostNameOverride];
1434  options.interceptorFactories = @[ [[DefaultInterceptorFactory alloc] init] ];
1435
1436  __block GRPCStreamingProtoCall *call = [_service
1437      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1438                                            initWithInitialMetadataCallback:nil
1439                                            messageCallback:^(id message) {
1440                                              XCTAssertLessThan(index, 4,
1441                                                                @"More than 4 responses received.");
1442                                              id expected = [RMTStreamingOutputCallResponse
1443                                                  messageWithPayloadSize:responses[index]];
1444                                              XCTAssertEqualObjects(message, expected);
1445                                              index += 1;
1446                                              if (index < 4) {
1447                                                id request = [RMTStreamingOutputCallRequest
1448                                                    messageWithPayloadSize:requests[index]
1449                                                     requestedResponseSize:responses[index]];
1450                                                [call writeMessage:request];
1451                                              } else {
1452                                                [call finish];
1453                                              }
1454                                            }
1455                                            closeCallback:^(NSDictionary *trailingMetadata,
1456                                                            NSError *error) {
1457                                              XCTAssertNil(error,
1458                                                           @"Finished with unexpected error: %@",
1459                                                           error);
1460                                              XCTAssertEqual(index, 4,
1461                                                             @"Received %i responses instead of 4.",
1462                                                             index);
1463                                              [expectation fulfill];
1464                                            }]
1465                            callOptions:options];
1466  [call start];
1467  [call writeMessage:request];
1468
1469  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1470}
1471
1472- (void)testLoggingInterceptor {
1473  XCTAssertNotNil([[self class] host]);
1474  __weak XCTestExpectation *expectation =
1475      [self expectationWithDescription:@"testLoggingInterceptor"];
1476
1477  __block NSUInteger startCount = 0;
1478  __block NSUInteger writeDataCount = 0;
1479  __block NSUInteger finishCount = 0;
1480  __block NSUInteger receiveNextMessageCount = 0;
1481  __block NSUInteger responseHeaderCount = 0;
1482  __block NSUInteger responseDataCount = 0;
1483  __block NSUInteger responseCloseCount = 0;
1484  __block NSUInteger didWriteDataCount = 0;
1485  id<GRPCInterceptorFactory> factory = [[HookInterceptorFactory alloc]
1486      initWithDispatchQueue:dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL)
1487      startHook:^(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
1488                  GRPCInterceptorManager *manager) {
1489        startCount++;
1490        XCTAssertEqualObjects(requestOptions.host, [[self class] host]);
1491        XCTAssertEqualObjects(requestOptions.path, @"/grpc.testing.TestService/FullDuplexCall");
1492        XCTAssertEqual(requestOptions.safety, GRPCCallSafetyDefault);
1493        [manager startNextInterceptorWithRequest:[requestOptions copy]
1494                                     callOptions:[callOptions copy]];
1495      }
1496      writeDataHook:^(id data, GRPCInterceptorManager *manager) {
1497        writeDataCount++;
1498        [manager writeNextInterceptorWithData:data];
1499      }
1500      finishHook:^(GRPCInterceptorManager *manager) {
1501        finishCount++;
1502        [manager finishNextInterceptor];
1503      }
1504      receiveNextMessagesHook:^(NSUInteger numberOfMessages, GRPCInterceptorManager *manager) {
1505        receiveNextMessageCount++;
1506        [manager receiveNextInterceptorMessages:numberOfMessages];
1507      }
1508      responseHeaderHook:^(NSDictionary *initialMetadata, GRPCInterceptorManager *manager) {
1509        responseHeaderCount++;
1510        [manager forwardPreviousInterceptorWithInitialMetadata:initialMetadata];
1511      }
1512      responseDataHook:^(id data, GRPCInterceptorManager *manager) {
1513        responseDataCount++;
1514        [manager forwardPreviousInterceptorWithData:data];
1515      }
1516      responseCloseHook:^(NSDictionary *trailingMetadata, NSError *error,
1517                          GRPCInterceptorManager *manager) {
1518        responseCloseCount++;
1519        [manager forwardPreviousInterceptorCloseWithTrailingMetadata:trailingMetadata error:error];
1520      }
1521      didWriteDataHook:^(GRPCInterceptorManager *manager) {
1522        didWriteDataCount++;
1523        [manager forwardPreviousInterceptorDidWriteData];
1524      }];
1525
1526  NSArray *requests = @[ @1, @2, @3, @4 ];
1527  NSArray *responses = @[ @1, @2, @3, @4 ];
1528
1529  __block int index = 0;
1530
1531  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1532                                               requestedResponseSize:responses[index]];
1533  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1534  // For backwards compatibility
1535  options.transportType = [[self class] transportType];
1536  options.transport = [[self class] transport];
1537  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1538  options.hostNameOverride = [[self class] hostNameOverride];
1539  options.flowControlEnabled = YES;
1540  options.interceptorFactories = @[ factory ];
1541  __block BOOL canWriteData = NO;
1542
1543  __block GRPCStreamingProtoCall *call = [_service
1544      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1545                                            initWithInitialMetadataCallback:nil
1546                                            messageCallback:^(id message) {
1547                                              XCTAssertLessThan(index, 4,
1548                                                                @"More than 4 responses received.");
1549                                              id expected = [RMTStreamingOutputCallResponse
1550                                                  messageWithPayloadSize:responses[index]];
1551                                              XCTAssertEqualObjects(message, expected);
1552                                              index += 1;
1553                                              if (index < 4) {
1554                                                id request = [RMTStreamingOutputCallRequest
1555                                                    messageWithPayloadSize:requests[index]
1556                                                     requestedResponseSize:responses[index]];
1557                                                XCTAssertTrue(canWriteData);
1558                                                canWriteData = NO;
1559                                                [call writeMessage:request];
1560                                                [call receiveNextMessage];
1561                                              } else {
1562                                                [call finish];
1563                                              }
1564                                            }
1565                                            closeCallback:^(NSDictionary *trailingMetadata,
1566                                                            NSError *error) {
1567                                              XCTAssertNil(error,
1568                                                           @"Finished with unexpected error: %@",
1569                                                           error);
1570                                              XCTAssertEqual(index, 4,
1571                                                             @"Received %i responses instead of 4.",
1572                                                             index);
1573                                              [expectation fulfill];
1574                                            }
1575                                            writeMessageCallback:^{
1576                                              canWriteData = YES;
1577                                            }]
1578                            callOptions:options];
1579  [call start];
1580  [call receiveNextMessage];
1581  [call writeMessage:request];
1582
1583  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1584  XCTAssertEqual(startCount, 1);
1585  XCTAssertEqual(writeDataCount, 4);
1586  XCTAssertEqual(finishCount, 1);
1587  XCTAssertEqual(receiveNextMessageCount, 4);
1588  XCTAssertEqual(responseHeaderCount, 1);
1589  XCTAssertEqual(responseDataCount, 4);
1590  XCTAssertEqual(responseCloseCount, 1);
1591  XCTAssertEqual(didWriteDataCount, 4);
1592}
1593
1594// Chain a default interceptor and a hook interceptor which, after two writes, cancels the call
1595// under the hood but forward further data to the user.
1596- (void)testHijackingInterceptor {
1597  NSUInteger kCancelAfterWrites = 2;
1598  XCTAssertNotNil([[self class] host]);
1599  __weak XCTestExpectation *expectUserCallComplete =
1600      [self expectationWithDescription:@"User call completed."];
1601  __weak XCTestExpectation *expectCallInternalComplete =
1602      [self expectationWithDescription:@"Internal gRPC call completed."];
1603
1604  NSArray *responses = @[ @1, @2, @3, @4 ];
1605  __block int index = 0;
1606
1607  __block NSUInteger startCount = 0;
1608  __block NSUInteger writeDataCount = 0;
1609  __block NSUInteger finishCount = 0;
1610  __block NSUInteger responseHeaderCount = 0;
1611  __block NSUInteger responseDataCount = 0;
1612  __block NSUInteger responseCloseCount = 0;
1613  id<GRPCInterceptorFactory> factory = [[HookInterceptorFactory alloc]
1614      initWithDispatchQueue:dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL)
1615      startHook:^(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
1616                  GRPCInterceptorManager *manager) {
1617        startCount++;
1618        [manager startNextInterceptorWithRequest:[requestOptions copy]
1619                                     callOptions:[callOptions copy]];
1620      }
1621      writeDataHook:^(id data, GRPCInterceptorManager *manager) {
1622        writeDataCount++;
1623        if (index < kCancelAfterWrites) {
1624          [manager writeNextInterceptorWithData:data];
1625        } else if (index == kCancelAfterWrites) {
1626          [manager cancelNextInterceptor];
1627          [manager forwardPreviousInterceptorWithData:[[RMTStreamingOutputCallResponse
1628                                                          messageWithPayloadSize:responses[index]]
1629                                                          data]];
1630        } else {  // (index > kCancelAfterWrites)
1631          [manager forwardPreviousInterceptorWithData:[[RMTStreamingOutputCallResponse
1632                                                          messageWithPayloadSize:responses[index]]
1633                                                          data]];
1634        }
1635      }
1636      finishHook:^(GRPCInterceptorManager *manager) {
1637        finishCount++;
1638        // finish must happen after the hijacking, so directly reply with a close
1639        [manager forwardPreviousInterceptorCloseWithTrailingMetadata:@{@"grpc-status" : @"0"}
1640                                                               error:nil];
1641        [manager shutDown];
1642      }
1643      receiveNextMessagesHook:nil
1644      responseHeaderHook:^(NSDictionary *initialMetadata, GRPCInterceptorManager *manager) {
1645        responseHeaderCount++;
1646        [manager forwardPreviousInterceptorWithInitialMetadata:initialMetadata];
1647      }
1648      responseDataHook:^(id data, GRPCInterceptorManager *manager) {
1649        responseDataCount++;
1650        [manager forwardPreviousInterceptorWithData:data];
1651      }
1652      responseCloseHook:^(NSDictionary *trailingMetadata, NSError *error,
1653                          GRPCInterceptorManager *manager) {
1654        responseCloseCount++;
1655        // since we canceled the call, it should return cancel error
1656        XCTAssertNil(trailingMetadata);
1657        XCTAssertNotNil(error);
1658        XCTAssertEqual(error.code, GRPC_STATUS_CANCELLED);
1659        [expectCallInternalComplete fulfill];
1660      }
1661      didWriteDataHook:nil];
1662
1663  NSArray *requests = @[ @1, @2, @3, @4 ];
1664
1665  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1666                                               requestedResponseSize:responses[index]];
1667  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1668  // For backwards compatibility
1669  options.transportType = [[self class] transportType];
1670  options.transport = [[self class] transport];
1671  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1672  options.hostNameOverride = [[self class] hostNameOverride];
1673  options.interceptorFactories = @[ [[DefaultInterceptorFactory alloc] init], factory ];
1674
1675  __block GRPCStreamingProtoCall *call = [_service
1676      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1677                                            initWithInitialMetadataCallback:nil
1678                                            messageCallback:^(id message) {
1679                                              XCTAssertLessThan(index, 4,
1680                                                                @"More than 4 responses received.");
1681                                              id expected = [RMTStreamingOutputCallResponse
1682                                                  messageWithPayloadSize:responses[index]];
1683                                              XCTAssertEqualObjects(message, expected);
1684                                              index += 1;
1685                                              if (index < 4) {
1686                                                id request = [RMTStreamingOutputCallRequest
1687                                                    messageWithPayloadSize:requests[index]
1688                                                     requestedResponseSize:responses[index]];
1689                                                [call writeMessage:request];
1690                                                [call receiveNextMessage];
1691                                              } else {
1692                                                [call finish];
1693                                              }
1694                                            }
1695                                            closeCallback:^(NSDictionary *trailingMetadata,
1696                                                            NSError *error) {
1697                                              XCTAssertNil(error,
1698                                                           @"Finished with unexpected error: %@",
1699                                                           error);
1700                                              XCTAssertEqual(index, 4,
1701                                                             @"Received %i responses instead of 4.",
1702                                                             index);
1703                                              [expectUserCallComplete fulfill];
1704                                            }]
1705                            callOptions:options];
1706  [call start];
1707  [call receiveNextMessage];
1708  [call writeMessage:request];
1709
1710  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1711  XCTAssertEqual(startCount, 1);
1712  XCTAssertEqual(writeDataCount, 4);
1713  XCTAssertEqual(finishCount, 1);
1714  XCTAssertEqual(responseHeaderCount, 1);
1715  XCTAssertEqual(responseDataCount, 2);
1716  XCTAssertEqual(responseCloseCount, 1);
1717}
1718
1719- (void)testGlobalInterceptor {
1720  XCTAssertNotNil([[self class] host]);
1721  __weak XCTestExpectation *expectation =
1722      [self expectationWithDescription:@"testGlobalInterceptor"];
1723
1724  __block NSUInteger startCount = 0;
1725  __block NSUInteger writeDataCount = 0;
1726  __block NSUInteger finishCount = 0;
1727  __block NSUInteger receiveNextMessageCount = 0;
1728  __block NSUInteger responseHeaderCount = 0;
1729  __block NSUInteger responseDataCount = 0;
1730  __block NSUInteger responseCloseCount = 0;
1731  __block NSUInteger didWriteDataCount = 0;
1732  [globalInterceptorFactory
1733      setStartHook:^(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
1734                     GRPCInterceptorManager *manager) {
1735        startCount++;
1736        XCTAssertEqualObjects(requestOptions.host, [[self class] host]);
1737        XCTAssertEqualObjects(requestOptions.path, @"/grpc.testing.TestService/FullDuplexCall");
1738        XCTAssertEqual(requestOptions.safety, GRPCCallSafetyDefault);
1739        [manager startNextInterceptorWithRequest:[requestOptions copy]
1740                                     callOptions:[callOptions copy]];
1741      }
1742      writeDataHook:^(id data, GRPCInterceptorManager *manager) {
1743        writeDataCount++;
1744        [manager writeNextInterceptorWithData:data];
1745      }
1746      finishHook:^(GRPCInterceptorManager *manager) {
1747        finishCount++;
1748        [manager finishNextInterceptor];
1749      }
1750      receiveNextMessagesHook:^(NSUInteger numberOfMessages, GRPCInterceptorManager *manager) {
1751        receiveNextMessageCount++;
1752        [manager receiveNextInterceptorMessages:numberOfMessages];
1753      }
1754      responseHeaderHook:^(NSDictionary *initialMetadata, GRPCInterceptorManager *manager) {
1755        responseHeaderCount++;
1756        [manager forwardPreviousInterceptorWithInitialMetadata:initialMetadata];
1757      }
1758      responseDataHook:^(id data, GRPCInterceptorManager *manager) {
1759        responseDataCount++;
1760        [manager forwardPreviousInterceptorWithData:data];
1761      }
1762      responseCloseHook:^(NSDictionary *trailingMetadata, NSError *error,
1763                          GRPCInterceptorManager *manager) {
1764        responseCloseCount++;
1765        [manager forwardPreviousInterceptorCloseWithTrailingMetadata:trailingMetadata error:error];
1766      }
1767      didWriteDataHook:^(GRPCInterceptorManager *manager) {
1768        didWriteDataCount++;
1769        [manager forwardPreviousInterceptorDidWriteData];
1770      }];
1771
1772  NSArray *requests = @[ @1, @2, @3, @4 ];
1773  NSArray *responses = @[ @1, @2, @3, @4 ];
1774
1775  __block int index = 0;
1776
1777  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1778                                               requestedResponseSize:responses[index]];
1779  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1780  // For backwards compatibility
1781  options.transportType = [[self class] transportType];
1782  options.transport = [[self class] transport];
1783  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1784  options.hostNameOverride = [[self class] hostNameOverride];
1785  options.flowControlEnabled = YES;
1786  globalInterceptorFactory.enabled = YES;
1787
1788  __block BOOL canWriteData = NO;
1789  __block GRPCStreamingProtoCall *call = [_service
1790      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1791                                            initWithInitialMetadataCallback:nil
1792                                            messageCallback:^(id message) {
1793                                              XCTAssertLessThan(index, 4,
1794                                                                @"More than 4 responses received.");
1795                                              index += 1;
1796                                              if (index < 4) {
1797                                                id request = [RMTStreamingOutputCallRequest
1798                                                    messageWithPayloadSize:requests[index]
1799                                                     requestedResponseSize:responses[index]];
1800                                                XCTAssertTrue(canWriteData);
1801                                                canWriteData = NO;
1802                                                [call writeMessage:request];
1803                                                [call receiveNextMessage];
1804                                              } else {
1805                                                [call finish];
1806                                              }
1807                                            }
1808                                            closeCallback:^(NSDictionary *trailingMetadata,
1809                                                            NSError *error) {
1810                                              XCTAssertNil(error,
1811                                                           @"Finished with unexpected error: %@",
1812                                                           error);
1813                                              [expectation fulfill];
1814                                            }
1815                                            writeMessageCallback:^{
1816                                              canWriteData = YES;
1817                                            }]
1818                            callOptions:options];
1819  [call start];
1820  [call receiveNextMessage];
1821  [call writeMessage:request];
1822
1823  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
1824  XCTAssertEqual(startCount, 1);
1825  XCTAssertEqual(writeDataCount, 4);
1826  XCTAssertEqual(finishCount, 1);
1827  XCTAssertEqual(receiveNextMessageCount, 4);
1828  XCTAssertEqual(responseHeaderCount, 1);
1829  XCTAssertEqual(responseDataCount, 4);
1830  XCTAssertEqual(responseCloseCount, 1);
1831  XCTAssertEqual(didWriteDataCount, 4);
1832  globalInterceptorFactory.enabled = NO;
1833}
1834
1835- (void)testConflictingGlobalInterceptors {
1836  id<GRPCInterceptorFactory> factory = [[HookInterceptorFactory alloc]
1837        initWithDispatchQueue:dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL)
1838                    startHook:nil
1839                writeDataHook:nil
1840                   finishHook:nil
1841      receiveNextMessagesHook:nil
1842           responseHeaderHook:nil
1843             responseDataHook:nil
1844            responseCloseHook:nil
1845             didWriteDataHook:nil];
1846  @try {
1847    [GRPCCall2 registerGlobalInterceptor:factory];
1848    XCTFail(@"Did not receive an exception when registering global interceptor the second time");
1849  } @catch (NSException *exception) {
1850    // Do nothing; test passes
1851  }
1852}
1853
1854- (void)testInterceptorAndGlobalInterceptor {
1855  XCTAssertNotNil([[self class] host]);
1856  __weak XCTestExpectation *expectation =
1857      [self expectationWithDescription:@"testInterceptorAndGlobalInterceptor"];
1858
1859  __block NSUInteger startCount = 0;
1860  __block NSUInteger writeDataCount = 0;
1861  __block NSUInteger finishCount = 0;
1862  __block NSUInteger receiveNextMessageCount = 0;
1863  __block NSUInteger responseHeaderCount = 0;
1864  __block NSUInteger responseDataCount = 0;
1865  __block NSUInteger responseCloseCount = 0;
1866  __block NSUInteger didWriteDataCount = 0;
1867
1868  id<GRPCInterceptorFactory> factory = [[HookInterceptorFactory alloc]
1869      initWithDispatchQueue:dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL)
1870      startHook:^(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
1871                  GRPCInterceptorManager *manager) {
1872        startCount++;
1873        XCTAssertEqualObjects(requestOptions.host, [[self class] host]);
1874        XCTAssertEqualObjects(requestOptions.path, @"/grpc.testing.TestService/FullDuplexCall");
1875        XCTAssertEqual(requestOptions.safety, GRPCCallSafetyDefault);
1876        [manager startNextInterceptorWithRequest:[requestOptions copy]
1877                                     callOptions:[callOptions copy]];
1878      }
1879      writeDataHook:^(id data, GRPCInterceptorManager *manager) {
1880        writeDataCount++;
1881        [manager writeNextInterceptorWithData:data];
1882      }
1883      finishHook:^(GRPCInterceptorManager *manager) {
1884        finishCount++;
1885        [manager finishNextInterceptor];
1886      }
1887      receiveNextMessagesHook:^(NSUInteger numberOfMessages, GRPCInterceptorManager *manager) {
1888        receiveNextMessageCount++;
1889        [manager receiveNextInterceptorMessages:numberOfMessages];
1890      }
1891      responseHeaderHook:^(NSDictionary *initialMetadata, GRPCInterceptorManager *manager) {
1892        responseHeaderCount++;
1893        [manager forwardPreviousInterceptorWithInitialMetadata:initialMetadata];
1894      }
1895      responseDataHook:^(id data, GRPCInterceptorManager *manager) {
1896        responseDataCount++;
1897        [manager forwardPreviousInterceptorWithData:data];
1898      }
1899      responseCloseHook:^(NSDictionary *trailingMetadata, NSError *error,
1900                          GRPCInterceptorManager *manager) {
1901        responseCloseCount++;
1902        [manager forwardPreviousInterceptorCloseWithTrailingMetadata:trailingMetadata error:error];
1903      }
1904      didWriteDataHook:^(GRPCInterceptorManager *manager) {
1905        didWriteDataCount++;
1906        [manager forwardPreviousInterceptorDidWriteData];
1907      }];
1908
1909  __block NSUInteger globalStartCount = 0;
1910  __block NSUInteger globalWriteDataCount = 0;
1911  __block NSUInteger globalFinishCount = 0;
1912  __block NSUInteger globalReceiveNextMessageCount = 0;
1913  __block NSUInteger globalResponseHeaderCount = 0;
1914  __block NSUInteger globalResponseDataCount = 0;
1915  __block NSUInteger globalResponseCloseCount = 0;
1916  __block NSUInteger globalDidWriteDataCount = 0;
1917
1918  [globalInterceptorFactory
1919      setStartHook:^(GRPCRequestOptions *requestOptions, GRPCCallOptions *callOptions,
1920                     GRPCInterceptorManager *manager) {
1921        globalStartCount++;
1922        XCTAssertEqualObjects(requestOptions.host, [[self class] host]);
1923        XCTAssertEqualObjects(requestOptions.path, @"/grpc.testing.TestService/FullDuplexCall");
1924        XCTAssertEqual(requestOptions.safety, GRPCCallSafetyDefault);
1925        [manager startNextInterceptorWithRequest:[requestOptions copy]
1926                                     callOptions:[callOptions copy]];
1927      }
1928      writeDataHook:^(id data, GRPCInterceptorManager *manager) {
1929        globalWriteDataCount++;
1930        [manager writeNextInterceptorWithData:data];
1931      }
1932      finishHook:^(GRPCInterceptorManager *manager) {
1933        globalFinishCount++;
1934        [manager finishNextInterceptor];
1935      }
1936      receiveNextMessagesHook:^(NSUInteger numberOfMessages, GRPCInterceptorManager *manager) {
1937        globalReceiveNextMessageCount++;
1938        [manager receiveNextInterceptorMessages:numberOfMessages];
1939      }
1940      responseHeaderHook:^(NSDictionary *initialMetadata, GRPCInterceptorManager *manager) {
1941        globalResponseHeaderCount++;
1942        [manager forwardPreviousInterceptorWithInitialMetadata:initialMetadata];
1943      }
1944      responseDataHook:^(id data, GRPCInterceptorManager *manager) {
1945        globalResponseDataCount++;
1946        [manager forwardPreviousInterceptorWithData:data];
1947      }
1948      responseCloseHook:^(NSDictionary *trailingMetadata, NSError *error,
1949                          GRPCInterceptorManager *manager) {
1950        globalResponseCloseCount++;
1951        [manager forwardPreviousInterceptorCloseWithTrailingMetadata:trailingMetadata error:error];
1952      }
1953      didWriteDataHook:^(GRPCInterceptorManager *manager) {
1954        globalDidWriteDataCount++;
1955        [manager forwardPreviousInterceptorDidWriteData];
1956      }];
1957
1958  NSArray *requests = @[ @1, @2, @3, @4 ];
1959  NSArray *responses = @[ @1, @2, @3, @4 ];
1960
1961  __block int index = 0;
1962
1963  id request = [RMTStreamingOutputCallRequest messageWithPayloadSize:requests[index]
1964                                               requestedResponseSize:responses[index]];
1965  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
1966  // For backwards compatibility
1967  options.transportType = [[self class] transportType];
1968  options.transport = [[self class] transport];
1969  options.PEMRootCertificates = [[self class] PEMRootCertificates];
1970  options.hostNameOverride = [[self class] hostNameOverride];
1971  options.flowControlEnabled = YES;
1972  options.interceptorFactories = @[ factory ];
1973  globalInterceptorFactory.enabled = YES;
1974
1975  __block BOOL canWriteData = NO;
1976  __block GRPCStreamingProtoCall *call = [_service
1977      fullDuplexCallWithResponseHandler:[[InteropTestsBlockCallbacks alloc]
1978                                            initWithInitialMetadataCallback:nil
1979                                            messageCallback:^(id message) {
1980                                              index += 1;
1981                                              if (index < 4) {
1982                                                id request = [RMTStreamingOutputCallRequest
1983                                                    messageWithPayloadSize:requests[index]
1984                                                     requestedResponseSize:responses[index]];
1985                                                canWriteData = NO;
1986                                                [call writeMessage:request];
1987                                                [call receiveNextMessage];
1988                                              } else {
1989                                                [call finish];
1990                                              }
1991                                            }
1992                                            closeCallback:^(NSDictionary *trailingMetadata,
1993                                                            NSError *error) {
1994                                              [expectation fulfill];
1995                                            }
1996                                            writeMessageCallback:^{
1997                                              canWriteData = YES;
1998                                            }]
1999                            callOptions:options];
2000  [call start];
2001  [call receiveNextMessage];
2002  [call writeMessage:request];
2003
2004  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
2005  XCTAssertEqual(startCount, 1);
2006  XCTAssertEqual(writeDataCount, 4);
2007  XCTAssertEqual(finishCount, 1);
2008  XCTAssertEqual(receiveNextMessageCount, 4);
2009  XCTAssertEqual(responseHeaderCount, 1);
2010  XCTAssertEqual(responseDataCount, 4);
2011  XCTAssertEqual(responseCloseCount, 1);
2012  XCTAssertEqual(didWriteDataCount, 4);
2013  XCTAssertEqual(globalStartCount, 1);
2014  XCTAssertEqual(globalWriteDataCount, 4);
2015  XCTAssertEqual(globalFinishCount, 1);
2016  XCTAssertEqual(globalReceiveNextMessageCount, 4);
2017  XCTAssertEqual(globalResponseHeaderCount, 1);
2018  XCTAssertEqual(globalResponseDataCount, 4);
2019  XCTAssertEqual(globalResponseCloseCount, 1);
2020  XCTAssertEqual(globalDidWriteDataCount, 4);
2021  globalInterceptorFactory.enabled = NO;
2022}
2023
2024@end
2025