Add sendMessages and flush to StreamingRequestClientCall

Sources/GRPC/ClientCalls/ClientCall.swift

@@ -76,6 +76,25 @@ public protocol StreamingRequestClientCall: ClientCall {
   ///   - flush: Whether the buffer should be flushed after writing the message.
   func sendMessage(_ message: RequestMessage, promise: EventLoopPromise<Void>?, flush: Bool)
 
+  /// Sends a sequence of messages to the service.
+  ///
+  /// - Important: Callers must terminate the stream of messages by calling `sendEnd()` or `sendEnd(promise:)`.
+  ///
+  /// - Parameters:
+  ///   - messages: The sequence of messages to send.
+  ///   - flush: Whether the buffer should be flushed after writing the message.
+  func sendMessages<S: Sequence>(_ messages: S, flush: Bool) -> EventLoopFuture<Void> where S.Element == RequestMessage
+
+  /// Sends a sequence of messages to the service.
+  ///
+  /// - Important: Callers must terminate the stream of messages by calling `sendEnd()` or `sendEnd(promise:)`.
+  ///
+  /// - Parameters:
+  ///   - messages: The sequence of messages to send.
+  ///   - promise: A promise to be fulfilled when all messages have been sent successfully.
+  ///   - flush: Whether the buffer should be flushed after writing the message.
+  func sendMessages<S: Sequence>(_ messages: S, promise: EventLoopPromise<Void>?, flush: Bool) where S.Element == RequestMessage
+
   /// Returns a future which can be used as a message queue.
   ///
   /// Callers may use this as such:
@@ -100,6 +119,9 @@ public protocol StreamingRequestClientCall: ClientCall {
   /// - Important: This should only ever be called once.
   /// - Parameter promise: A promise to be fulfilled when the end has been sent.
   func sendEnd(promise: EventLoopPromise<Void>?)
+
+  /// Flush the buffer of messages.
+  func flush()
 }
 
 /// A `ClientCall` with a unary response; i.e. unary and client-streaming.
@@ -131,6 +153,33 @@ extension StreamingRequestClientCall {
     }
   }
 
+  public func sendMessages<S: Sequence>(_ messages: S, flush: Bool = true) -> EventLoopFuture<Void> where S.Element == RequestMessage {
+    self.subchannel.flatMap { channel in
+      let writeFutures = messages.map { message in
+        channel.write(GRPCClientRequestPart.message(_Box(message)))
+      }
+      if flush {
+        channel.flush()
+      }
+      return EventLoopFuture.andAllSucceed(writeFutures, on: channel.eventLoop)
+    }
+  }
+
+  public func sendMessages<S: Sequence>(_ messages: S, promise: EventLoopPromise<Void>?, flush: Bool = true) where S.Element == RequestMessage {
+    if let promise = promise {
+      self.sendMessages(messages, flush: flush).cascade(to: promise)
+    } else {
+      self.subchannel.whenSuccess { channel in
+        for message in messages {
+          channel.write(GRPCClientRequestPart.message(_Box(message)), promise: nil)
+        }
+        if flush {
+          channel.flush()
+        }
+      }
+    }
+  }
+
   public func sendEnd() -> EventLoopFuture<Void> {
     return self.subchannel.flatMap { channel in
       return channel.writeAndFlush(GRPCClientRequestPart<RequestMessage>.end)
@@ -142,4 +191,10 @@ extension StreamingRequestClientCall {
       channel.writeAndFlush(GRPCClientRequestPart<RequestMessage>.end, promise: promise)
     }
   }
+
+  public func flush() {
+    self.subchannel.whenSuccess { channel in
+      channel.flush()
+    }
+  }
 }

Tests/GRPCTests/StreamingRequestClientCallTests.swift

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2019, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import Foundation
+import GRPC
+import XCTest
+
+class StreamingRequestClientCallTests: EchoTestCaseBase {
+  class ResponseCounter {
+    var expectation: XCTestExpectation
+
+    init(expectation: XCTestExpectation) {
+      self.expectation = expectation
+    }
+
+    func increment() {
+      self.expectation.fulfill()
+    }
+  }
+
+  func testSendMessages() throws {
+    let firstBatchReceived = self.expectation(description: "first batch received")
+    let counter = ResponseCounter(expectation: firstBatchReceived)
+
+    let update = self.client.update { _ in
+      counter.increment()
+    }
+
+    // Send the first batch.
+    let firstBatch = ["foo", "bar", "baz"].map { Echo_EchoRequest(text: $0) }
+    firstBatchReceived.expectedFulfillmentCount = firstBatch.count
+    XCTAssertNoThrow(try update.sendMessages(firstBatch).wait())
+
+    // Wait for the first batch of resonses.
+    self.wait(for: [firstBatchReceived], timeout: 0.5)
+
+    // Send more messages, but don't flush.
+    let secondBatchNotReceived = self.expectation(description: "second batch not received")
+    secondBatchNotReceived.isInverted = true
+    counter.expectation = secondBatchNotReceived
+
+    let secondBatch = (0..<3).map { Echo_EchoRequest(text: "\($0)") }
+    update.sendMessages(secondBatch, promise: nil, flush: false)
+
+    // Wait and check that the expectation hasn't been fulfilled (because we haven't flushed).
+    self.wait(for: [secondBatchNotReceived], timeout: 0.5)
+
+    let secondBatchReceived = self.expectation(description: "second batch received")
+    secondBatchReceived.expectedFulfillmentCount = secondBatch.count
+    counter.expectation = secondBatchReceived
+
+    // Flush the messages: we should get responses now.
+    update.flush()
+    self.wait(for: [secondBatchReceived], timeout: 0.5)
+
+    // End the call.
+    update.sendEnd(promise: nil)
+
+    let statusReceived = self.expectation(description: "status received")
+    update.status.map { $0.code }.assertEqual(.ok, fulfill: statusReceived)
+
+    self.wait(for: [statusReceived], timeout: 1.0)
+  }
+
+}

Tests/GRPCTests/XCTestManifests.swift

@@ -433,6 +433,15 @@ extension ServerWebTests {
     ]
 }
 
+extension StreamingRequestClientCallTests {
+    // DO NOT MODIFY: This is autogenerated, use:
+    //   `swift test --generate-linuxmain`
+    // to regenerate.
+    static let __allTests__StreamingRequestClientCallTests = [
+        ("testSendMessages", testSendMessages),
+    ]
+}
+
 public func __allTests() -> [XCTestCaseEntry] {
     return [
         testCase(AnyServiceClientTests.__allTests__AnyServiceClientTests),
@@ -464,6 +473,7 @@ public func __allTests() -> [XCTestCaseEntry] {
         testCase(ServerErrorTransformingTests.__allTests__ServerErrorTransformingTests),
         testCase(ServerThrowingTests.__allTests__ServerThrowingTests),
         testCase(ServerWebTests.__allTests__ServerWebTests),
+        testCase(StreamingRequestClientCallTests.__allTests__StreamingRequestClientCallTests),
     ]
 }
 #endif