Add a blocking variant of the `send` method to `ServerStreaming`.

Sources/Examples/Echo/Generated/echo.grpc.swift

@@ -47,8 +47,10 @@ class Echo_EchoExpandCallTestStub: ClientCallServerStreamingTestStub<Echo_EchoRe
 }
 
 internal protocol Echo_EchoCollectCall: ClientCallClientStreaming {
-  /// Call this to send each message in the request stream. Nonblocking.
+  /// Send a message to the stream. Nonblocking.
   func send(_ message: Echo_EchoRequest, completion: @escaping (Error?) -> Void) throws
+  /// Send a message to the stream and wait for the send operation to finish. Blocking.
+  func send(_ message: Echo_EchoRequest) throws
 
   /// Call this to close the connection and wait for a response. Blocking.
   func closeAndReceive() throws -> Echo_EchoResponse
@@ -72,8 +74,10 @@ internal protocol Echo_EchoUpdateCall: ClientCallBidirectionalStreaming {
   /// Call this to wait for a result. Nonblocking.
   func receive(completion: @escaping (ResultOrRPCError<Echo_EchoResponse?>) -> Void) throws
 
-  /// Call this to send each message in the request stream. Nonblocking.
+  /// Send a message to the stream. Nonblocking.
   func send(_ message: Echo_EchoRequest, completion: @escaping (Error?) -> Void) throws
+  /// Send a message to the stream and wait for the send operation to finish. Blocking.
+  func send(_ message: Echo_EchoRequest) throws
 
   /// Call this to close the sending connection. Blocking.
   func closeSend() throws
@@ -201,8 +205,10 @@ fileprivate final class Echo_EchoGetSessionBase: ServerSessionUnaryBase<Echo_Ech
 class Echo_EchoGetSessionTestStub: ServerSessionUnaryTestStub, Echo_EchoGetSession {}
 
 internal protocol Echo_EchoExpandSession: ServerSessionServerStreaming {
-  /// Call this to send each message in the request stream. Nonblocking.
+  /// Send a message to the stream. Nonblocking.
   func send(_ message: Echo_EchoResponse, completion: @escaping (Error?) -> Void) throws
+  /// Send a message to the stream and wait for the send operation to finish. Blocking.
+  func send(_ message: Echo_EchoResponse) throws
 
   /// Close the connection and send the status. Non-blocking.
   /// You MUST call this method once you are done processing the request.
@@ -238,8 +244,10 @@ internal protocol Echo_EchoUpdateSession: ServerSessionBidirectionalStreaming {
   /// Call this to wait for a result. Nonblocking.
   func receive(completion: @escaping (ResultOrRPCError<Echo_EchoRequest?>) -> Void) throws
 
-  /// Call this to send each message in the request stream. Nonblocking.
+  /// Send a message to the stream. Nonblocking.
   func send(_ message: Echo_EchoResponse, completion: @escaping (Error?) -> Void) throws
+  /// Send a message to the stream and wait for the send operation to finish. Blocking.
+  func send(_ message: Echo_EchoResponse) throws
 
   /// Close the connection and send the status. Non-blocking.
   /// You MUST call this method once you are done processing the request.

Sources/SwiftGRPC/Runtime/ClientCallBidirectionalStreaming.swift

@@ -71,6 +71,10 @@ open class ClientCallBidirectionalStreamingTestStub<InputType: Message, OutputTy
   open func send(_ message: InputType, completion _: @escaping (Error?) -> Void) throws {
     inputs.append(message)
   }
+  
+  open func send(_ message: InputType) throws {
+    inputs.append(message)
+  }
 
   open func closeSend(completion: (() -> Void)?) throws { completion?() }
 

Sources/SwiftGRPC/Runtime/ClientCallClientStreaming.swift

@@ -75,6 +75,10 @@ open class ClientCallClientStreamingTestStub<InputType: Message, OutputType: Mes
   open func send(_ message: InputType, completion _: @escaping (Error?) -> Void) throws {
     inputs.append(message)
   }
+  
+  open func send(_ message: InputType) throws {
+    inputs.append(message)
+  }
 
   open func closeAndReceive(completion: @escaping (ResultOrRPCError<OutputType>) -> Void) throws {
     completion(.result(output!))

Sources/SwiftGRPC/Runtime/ServerSessionBidirectionalStreaming.swift

@@ -82,6 +82,10 @@ open class ServerSessionBidirectionalStreamingTestStub<InputType: Message, Outpu
     outputs.append(message)
   }
 
+  open func send(_ message: OutputType) throws {
+    outputs.append(message)
+  }
+
   open func close(withStatus status: ServerStatus, completion: ((CallResult) -> Void)?) throws {
     self.status = status
     completion?(.fakeOK)

Sources/SwiftGRPC/Runtime/ServerSessionServerStreaming.swift

@@ -72,6 +72,10 @@ open class ServerSessionServerStreamingTestStub<OutputType: Message>: ServerSess
     outputs.append(message)
   }
 
+  open func send(_ message: OutputType) throws {
+    outputs.append(message)
+  }
+
   open func close(withStatus status: ServerStatus, completion: ((CallResult) -> Void)?) throws {
     self.status = status
     completion?(.fakeOK)

Sources/SwiftGRPC/Runtime/StreamSending.swift

@@ -29,6 +29,19 @@ extension StreamSending {
     try call.sendMessage(data: message.serializedData(), completion: completion)
   }
   
+  public func send(_ message: SentType) throws {
+    var resultError: Error?
+    let sem = DispatchSemaphore(value: 0)
+    try send(message) {
+      resultError = $0
+      sem.signal()
+    }
+    _ = sem.wait()
+    if let resultError = resultError {
+      throw resultError
+    }
+  }
+  
   public func waitForSendOperationsToFinish() {
     call.messageQueueEmpty.wait()
   }

Sources/protoc-gen-swiftgrpc/Generator-Methods.swift

@@ -26,7 +26,9 @@ extension Generator {
   }
   
   func printStreamSendMethods(sentType: String) {
-    println("/// Call this to send each message in the request stream. Nonblocking.")
+    println("/// Send a message to the stream. Nonblocking.")
     println("func send(_ message: \(sentType), completion: @escaping (Error?) -> Void) throws")
+    println("/// Send a message to the stream and wait for the send operation to finish. Blocking.")
+    println("func send(_ message: \(sentType)) throws")
   }
 }