Adding flow control support API for GRPCUnaryProtoCall (#26969)

This PR added receiveMessage(s) API on GRPCUnaryProtoCall for manually receiving messasges when flow control is enabled:
   * Implementation for GRPCUnaryProtoCall:receiveMessage(s) by chaining calls to GRPCStreamingProtoCall instance
   * Added finish API on GRPCUnaryProtoCall
   * On [GRPCUnaryProtoCall start], avoid closing streaming RPC if flow control is enabled.
   * internal call to receiveMessage is removed since message is automatically received if flow control not enabled.

3 Unit tests added to InteropTests suite for validating unary call API with flow control
 - testUnaryRPCWithV2APIFlowControl
 - testUnaryRPCWithV2APIFlowControlNotReceivingMessage
 - testPingPongUnaryRPCWithFlowControl

src/objective-c/ProtoRPC/ProtoRPC.h

@@ -126,6 +126,33 @@ NS_ASSUME_NONNULL_BEGIN
  */
 - (void)cancel;
 
+/**
+ * Finish the RPC request and half-close the call. The server may still send messages and/or
+ * trailers to the client. This method should only be used when flow control is enabled. If flow
+ * control is not enabled, It will be automatically called upon message received.
+ */
+- (void)finish;
+
+/**
+ * Tell gRPC to receive another message.
+ *
+ * This method should only be used when flow control is enabled. If flow control is enabled, gRPC
+ * will only receive additional messages after the user indicates so by using either
+ * receiveNextMessage: or receiveNextMessages: methods. If flow control is not enabled, messages
+ * will be automatically received after the previous one is delivered.
+ */
+- (void)receiveNextMessage;
+
+/**
+ * Tell gRPC to receive another N message.
+ *
+ * This method should only be used when flow control is enabled. If flow control is enabled, the
+ * messages received from the server are buffered in gRPC until the user want to receive the next
+ * message. If flow control is not enabled, messages will be automatically received after the
+ * previous one is delivered.
+ */
+- (void)receiveNextMessages:(NSUInteger)numberOfMessages;
+
 @end
 
 /** A client-streaming RPC call with Protobuf. */

src/objective-c/ProtoRPC/ProtoRPC.m

@@ -27,6 +27,14 @@
 #import <RxLibrary/GRXWriteable.h>
 #import <RxLibrary/GRXWriter+Transformations.h>
 
+#pragma mark - GRPCStreamingProtoCall
+
+@interface GRPCStreamingProtoCall () <GRPCResponseHandler>
+
+@property(nonatomic, readonly) GRPCCallOptions *callOptions;
+
+@end
+
 @implementation GRPCUnaryResponseHandler {
   void (^_responseHandler)(id, NSError *);
   dispatch_queue_t _responseDispatchQueue;
@@ -100,25 +108,35 @@
 
 - (void)start {
   [_call start];
-  [_call receiveNextMessage];
   [_call writeMessage:_message];
-  [_call finish];
+
+  if (!_call.callOptions.flowControlEnabled) {
+    [_call finish];
+  }
 }
 
 - (void)cancel {
   [_call cancel];
 }
 
-@end
+- (void)finish {
+  [_call finish];
+}
 
-@interface GRPCStreamingProtoCall () <GRPCResponseHandler>
+#pragma mark - GRPCControllableProtoCallFlow
+
+- (void)receiveNextMessage {
+  [_call receiveNextMessage];
+}
+- (void)receiveNextMessages:(NSUInteger)numberOfMessages {
+  [_call receiveNextMessages:numberOfMessages];
+}
 
 @end
 
 @implementation GRPCStreamingProtoCall {
   GRPCRequestOptions *_requestOptions;
   id<GRPCProtoResponseHandler> _handler;
-  GRPCCallOptions *_callOptions;
   Class _responseClass;
 
   GRPCCall2 *_call;
@@ -226,6 +244,8 @@
   [copiedCall finish];
 }
 
+#pragma mark - GRPCControllableProtoCallFlow
+
 - (void)receiveNextMessage {
   [self receiveNextMessages:1];
 }
@@ -237,6 +257,8 @@
   [copiedCall receiveNextMessages:numberOfMessages];
 }
 
+#pragma mark - GRPCResponseHandler
+
 - (void)didReceiveInitialMetadata:(NSDictionary *)initialMetadata {
   @synchronized(self) {
     if (initialMetadata != nil &&
@@ -314,6 +336,8 @@
   }
 }
 
+#pragma mark - GRPCDispatchable
+
 - (dispatch_queue_t)dispatchQueue {
   return _dispatchQueue;
 }

src/objective-c/tests/InteropTests/InteropTests.m

@@ -701,6 +701,87 @@ static dispatch_once_t initGlobalInterceptorFactory;
   [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
 }
 
+- (void)testUnaryRPCWithV2APIFlowControl {
+  XCTAssertNotNil([[self class] host]);
+  __weak XCTestExpectation *expectReceive =
+      [self expectationWithDescription:@"LargeUnaryWithV2API received message"];
+  __weak XCTestExpectation *expectComplete =
+      [self expectationWithDescription:@"LargeUnaryWithV2API received complete"];
+
+  const int responseSize = 123;
+  RMTSimpleRequest *request = [RMTSimpleRequest message];
+  request.responseType = RMTPayloadType_Compressable;
+  request.responseSize = responseSize;
+  request.payload.body = [NSMutableData dataWithLength:456];
+
+  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
+  // For backwards compatibility
+  options.transportType = [[self class] transportType];
+  options.transport = [[self class] transport];
+  options.PEMRootCertificates = [[self class] PEMRootCertificates];
+  options.hostNameOverride = [[self class] hostNameOverride];
+  options.flowControlEnabled = YES;
+
+  GRPCUnaryProtoCall *call = [_service
+      unaryCallWithMessage:request
+           responseHandler:[[InteropTestsBlockCallbacks alloc] initWithInitialMetadataCallback:nil
+                               messageCallback:^(id message) {
+                                 XCTAssertNotNil(message);
+                                 if (message) {
+                                   RMTSimpleResponse *expectedResponse =
+                                       [RMTSimpleResponse message];
+                                   expectedResponse.payload.type = RMTPayloadType_Compressable;
+                                   expectedResponse.payload.body =
+                                       [NSMutableData dataWithLength:responseSize];
+                                   XCTAssertEqualObjects(message, expectedResponse);
+                                   [expectReceive fulfill];
+                                   [call finish];
+                                 }
+                               }
+                               closeCallback:^(NSDictionary *trailingMetadata, NSError *error) {
+                                 XCTAssertNil(error, @"Unexpected error: %@", error);
+                                 [expectComplete fulfill];
+                               }]
+               callOptions:options];
+  [call start];
+  [call receiveNextMessage];
+  [self waitForExpectationsWithTimeout:TEST_TIMEOUT handler:nil];
+}
+
+- (void)testUnaryRPCWithV2APIFlowControlNotReceivingMessage {
+  XCTAssertNotNil([[self class] host]);
+
+  __weak XCTestExpectation *expectTimeout =
+      [self expectationWithDescription:
+                @"testUnaryRPCWithV2APIFlowControlNotReceivingMessage received timeout"];
+
+  RMTSimpleRequest *request = [RMTSimpleRequest message];
+  request.responseType = RMTPayloadType_Compressable;
+  request.responseSize = 123;
+  request.payload.body = [NSMutableData dataWithLength:456];
+
+  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
+  // For backwards compatibility
+  options.transportType = [[self class] transportType];
+  options.transport = [[self class] transport];
+  options.PEMRootCertificates = [[self class] PEMRootCertificates];
+  options.hostNameOverride = [[self class] hostNameOverride];
+  options.flowControlEnabled = YES;
+
+  GRPCUnaryProtoCall *call = [_service
+      unaryCallWithMessage:request
+           responseHandler:[[InteropTestsBlockCallbacks alloc]
+                               initWithInitialMetadataCallback:nil
+                                               messageCallback:^(id message) {
+                                                 XCTFail("Not expected to receive message");
+                                               }
+                                                 closeCallback:nil]
+               callOptions:options];
+  [call start];
+  XCTWaiterResult result = [XCTWaiter waitForExpectations:@[ expectTimeout ] timeout:5];
+  XCTAssertEqual(XCTWaiterResultTimedOut, result, @"Unexpected waiter result %@", @(result));
+}
+
 - (void)testConcurrentRPCsWithErrorsWithV2API {
   NSMutableArray *completeExpectations = [NSMutableArray array];
   NSMutableArray *calls = [NSMutableArray array];
@@ -1078,9 +1159,72 @@ static dispatch_once_t initGlobalInterceptorFactory;
   [self waitForExpectationsWithTimeout:STREAMING_CALL_TEST_TIMEOUT handler:nil];
 }
 
-- (void)testPingPongRPCWithFlowControl {
+- (void)testPingPongUnaryRPCWithFlowControl {
   XCTAssertNotNil([[self class] host]);
-  __weak XCTestExpectation *expectation = [self expectationWithDescription:@"PingPongWithV2API"];
+  __weak XCTestExpectation *expectation =
+      [self expectationWithDescription:@"UnaryPingPongWithV2API"];
+
+  NSNumber *requestSize = @321;
+  NSArray *responseSizes = @[ @123, @234 ];
+
+  RMTStreamingOutputCallRequest *request = [RMTStreamingOutputCallRequest message];
+  request.payload.body = [NSMutableData dataWithLength:requestSize.unsignedIntegerValue];
+  for (NSNumber *responseSize in responseSizes) {
+    RMTResponseParameters *parameters = [RMTResponseParameters message];
+    parameters.size = responseSize.intValue;
+    [request.responseParametersArray addObject:parameters];
+  }
+
+  GRPCMutableCallOptions *options = [[GRPCMutableCallOptions alloc] init];
+  // For backwards compatibility
+  options.transportType = [[self class] transportType];
+  options.transport = [[self class] transport];
+  options.PEMRootCertificates = [[self class] PEMRootCertificates];
+  options.hostNameOverride = [[self class] hostNameOverride];
+  options.flowControlEnabled = YES;
+
+  __block GRPCUnaryProtoCall *call = nil;
+  __block int receivedMessageCount = 0;
+
+  id messageHandler = ^(id message) {
+    NSLog(@"received message %@", @(receivedMessageCount));
+    XCTAssertLessThan(receivedMessageCount, responseSizes.count,
+                      "More than expected messages received");
+    id expected =
+        [RMTStreamingOutputCallResponse messageWithPayloadSize:responseSizes[receivedMessageCount]];
+    XCTAssertEqualObjects(message, expected);
+
+    receivedMessageCount += 1;
+    if (receivedMessageCount < responseSizes.count) {
+      [call receiveNextMessage];
+    } else {
+      [call finish];
+    }
+  };
+
+  id closeHandler = ^(NSDictionary *trailingMetadata, NSError *error) {
+    XCTAssertNil(error, @"Finished with unexpected error: %@", error);
+    [expectation fulfill];
+  };
+
+  InteropTestsBlockCallbacks *handler =
+      [[InteropTestsBlockCallbacks alloc] initWithInitialMetadataCallback:nil
+                                                          messageCallback:messageHandler
+                                                            closeCallback:closeHandler
+                                                     writeMessageCallback:nil];
+
+  call = [_service streamingOutputCallWithMessage:request
+                                  responseHandler:handler
+                                      callOptions:options];
+  [call start];
+  [call receiveNextMessage];
+  [self waitForExpectationsWithTimeout:STREAMING_CALL_TEST_TIMEOUT handler:nil];
+}
+
+- (void)testPingPongStreamingRPCWithFlowControl {
+  XCTAssertNotNil([[self class] host]);
+  __weak XCTestExpectation *expectation =
+      [self expectationWithDescription:@"StreamingPingPongWithV2API"];
 
   NSArray *requests = @[ @27182, @8, @1828, @45904 ];
   NSArray *responses = @[ @31415, @9, @2653, @58979 ];