Add a runtime support class for server-streaming client calls.

Examples/Echo/Generated/echo.grpc.swift

@@ -39,94 +39,20 @@ fileprivate final class Echo_EchoGetCallImpl: ClientCallUnaryImpl<Echo_EchoReque
   override class var method: String { return "/echo.Echo/Get" }
 }
 
-/// Expand (Server Streaming)
-internal protocol Echo_EchoExpandCall {
+internal protocol Echo_EchoExpandCall: ClientCallServerStreamingBase {
   /// Call this to wait for a result. Blocking.
   func receive() throws -> Echo_EchoResponse
   /// Call this to wait for a result. Nonblocking.
-  func receive(completion:@escaping (Echo_EchoResponse?, Echo_EchoClientError?)->()) throws
-  
-  /// Cancel the call.
-  func cancel()
+  func receive(completion:@escaping (Echo_EchoResponse?, ClientError?)->()) throws
 }
 
-internal extension Echo_EchoExpandCall {
-  func receive() throws -> Echo_EchoResponse {
-    var returnError : Echo_EchoClientError?
-    var returnResponse : Echo_EchoResponse!
-    let sem = DispatchSemaphore(value: 0)
-    do {
-      try receive() {response, error in
-        returnResponse = response
-        returnError = error
-        sem.signal()
-      }
-      _ = sem.wait(timeout: DispatchTime.distantFuture)
-    }
-    if let returnError = returnError {
-      throw returnError
-    }
-    return returnResponse
-  }
-}
-
-fileprivate final class Echo_EchoExpandCallImpl: Echo_EchoExpandCall {
-  private var call : Call
-
-  /// Create a call.
-  init(_ channel: Channel) {
-    self.call = channel.makeCall("/echo.Echo/Expand")
-  }
-
-  /// Call this once with the message to send. Nonblocking.
-  func start(request: Echo_EchoRequest,
-                         metadata: Metadata,
-                         completion: ((CallResult) -> ())?)
-    throws -> Echo_EchoExpandCall {
-      let requestData = try request.serializedData()
-      try call.start(.serverStreaming,
-                     metadata:metadata,
-                     message:requestData,
-                     completion:completion)
-      return self
-  }
-
-  func receive(completion:@escaping (Echo_EchoResponse?, Echo_EchoClientError?)->()) throws {
-    do {
-      try call.receiveMessage() {(responseData) in
-        if let responseData = responseData {
-          if let response = try? Echo_EchoResponse(serializedData:responseData) {
-            completion(response, nil)
-          } else {
-            completion(nil, Echo_EchoClientError.invalidMessageReceived)
-          }
-        } else {
-          completion(nil, Echo_EchoClientError.endOfStream)
-        }
-      }
-    }
-  }
-
-  /// Cancel the call.
-  func cancel() {
-    call.cancel()
-  }
+fileprivate final class Echo_EchoExpandCallImpl: ClientCallServerStreamingImpl<Echo_EchoRequest, Echo_EchoResponse>, Echo_EchoExpandCall {
+  override class var method: String { return "/echo.Echo/Expand" }
 }
 
 /// Simple fake implementation of Echo_EchoExpandCall that returns a previously-defined set of results.
-class Echo_EchoExpandCallTestStub: Echo_EchoExpandCall {
-  var outputs: [Echo_EchoResponse] = []
-  
-  func receive(completion:@escaping (Echo_EchoResponse?, Echo_EchoClientError?)->()) throws {
-    if let output = outputs.first {
-      outputs.removeFirst()
-      completion(output, nil)
-    } else {
-      completion(nil, Echo_EchoClientError.endOfStream)
-    }
-  }
-
-  func cancel() { }
+class Echo_EchoExpandCallTestStub: ClientCallServerStreamingTestStub<Echo_EchoResponse>, Echo_EchoExpandCall {
+  override class var method: String { return "/echo.Echo/Expand" }
 }
 
 /// Collect (Client Streaming)

Plugin/Templates/client-call-serverstreaming.swift

@@ -1,91 +1,16 @@
-/// {{ method|methodDescriptorName }} (Server Streaming)
-{{ access }} protocol {{ .|call:file,service,method }} {
+{{ access }} protocol {{ .|call:file,service,method }}: ClientCallServerStreamingBase {
   /// Call this to wait for a result. Blocking.
   func receive() throws -> {{ method|output }}
   /// Call this to wait for a result. Nonblocking.
-  func receive(completion:@escaping ({{ method|output }}?, {{ .|clienterror:file,service }}?)->()) throws
-  
-  /// Cancel the call.
-  func cancel()
+  func receive(completion: @escaping ({{ method|output }}?, ClientError?) -> Void) throws
 }
 
-{{ access }} extension {{ .|call:file,service,method }} {
-  func receive() throws -> {{ method|output }} {
-    var returnError : {{ .|clienterror:file,service }}?
-    var returnResponse : {{ method|output }}!
-    let sem = DispatchSemaphore(value: 0)
-    do {
-      try receive() {response, error in
-        returnResponse = response
-        returnError = error
-        sem.signal()
-      }
-      _ = sem.wait(timeout: DispatchTime.distantFuture)
-    }
-    if let returnError = returnError {
-      throw returnError
-    }
-    return returnResponse
-  }
-}
-
-fileprivate final class {{ .|call:file,service,method }}Impl: {{ .|call:file,service,method }} {
-  private var call : Call
-
-  /// Create a call.
-  init(_ channel: Channel) {
-    self.call = channel.makeCall("{{ .|path:file,service,method }}")
-  }
-
-  /// Call this once with the message to send. Nonblocking.
-  func start(request: {{ method|input }},
-                         metadata: Metadata,
-                         completion: ((CallResult) -> ())?)
-    throws -> {{ .|call:file,service,method }} {
-      let requestData = try request.serializedData()
-      try call.start(.serverStreaming,
-                     metadata:metadata,
-                     message:requestData,
-                     completion:completion)
-      return self
-  }
-
-  func receive(completion:@escaping ({{ method|output }}?, {{ .|clienterror:file,service }}?)->()) throws {
-    do {
-      try call.receiveMessage() {(responseData) in
-        if let responseData = responseData {
-          if let response = try? {{ method|output }}(serializedData:responseData) {
-            completion(response, nil)
-          } else {
-            completion(nil, {{ .|clienterror:file,service }}.invalidMessageReceived)
-          }
-        } else {
-          completion(nil, {{ .|clienterror:file,service }}.endOfStream)
-        }
-      }
-    }
-  }
-
-  /// Cancel the call.
-  func cancel() {
-    call.cancel()
-  }
+fileprivate final class {{ .|call:file,service,method }}Impl: ClientCallServerStreamingImpl<{{ method|input }}, {{ method|output }}>, {{ .|call:file,service,method }} {
+  override class var method: String { return "{{ .|path:file,service,method }}" }
 }
 
 //-{% if generateTestStubs %}
-/// Simple fake implementation of {{ .|call:file,service,method }} that returns a previously-defined set of results.
-class {{ .|call:file,service,method }}TestStub: {{ .|call:file,service,method }} {
-  var outputs: [{{ method|output }}] = []
-  
-  func receive(completion:@escaping ({{ method|output }}?, {{ .|clienterror:file,service }}?)->()) throws {
-    if let output = outputs.first {
-      outputs.removeFirst()
-      completion(output, nil)
-    } else {
-      completion(nil, {{ .|clienterror:file,service }}.endOfStream)
-    }
-  }
-
-  func cancel() { }
+class {{ .|call:file,service,method }}TestStub: ClientCallServerStreamingTestStub<{{ method|output }}>, {{ .|call:file,service,method }} {
+  override class var method: String { return "{{ .|path:file,service,method }}" }
 }
 //-{% endif %}

Sources/gRPC/GenCodeSupport/ClientCallServerStreaming.swift

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2018, 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 Dispatch
+import SwiftProtobuf
+
+public protocol ClientCallServerStreamingBase: class {
+  static var method: String { get }
+  
+  /// Cancel the call.
+  func cancel()
+  
+  // TODO: Move the other, message type-dependent, methods into this protocol. At the moment, this is not possible,
+  // as the protocol would then have an associated type requirement (and become pretty much unusable in the process).
+}
+
+open class ClientCallServerStreamingImpl<InputType: Message, OutputType: Message>: ClientCallServerStreamingBase {
+  open class var method: String { fatalError("needs to be overridden") }
+  
+  private var call: Call
+  
+  /// Create a call.
+  public init(_ channel: Channel) {
+    self.call = channel.makeCall(type(of: self).method)
+  }
+  
+  /// Call this once with the message to send. Nonblocking.
+  public func start(request: InputType, metadata: Metadata, completion: ((CallResult) -> ())?) throws -> Self {
+    let requestData = try request.serializedData()
+    try call.start(.serverStreaming,
+                   metadata:metadata,
+                   message:requestData,
+                   completion:completion)
+    return self
+  }
+  
+  public func receive(completion:@escaping (OutputType?, ClientError?)->()) throws {
+    do {
+      try call.receiveMessage() {(responseData) in
+        if let responseData = responseData {
+          if let response = try? OutputType(serializedData:responseData) {
+            completion(response, nil)
+          } else {
+            completion(nil, .invalidMessageReceived)
+          }
+        } else {
+          completion(nil, .endOfStream)
+        }
+      }
+    }
+  }
+  
+  public func receive() throws -> OutputType {
+    var returnError : ClientError?
+    var returnResponse : OutputType!
+    let sem = DispatchSemaphore(value: 0)
+    do {
+      try receive() {response, error in
+        returnResponse = response
+        returnError = error
+        sem.signal()
+      }
+      _ = sem.wait(timeout: DispatchTime.distantFuture)
+    }
+    if let returnError = returnError {
+      throw returnError
+    }
+    return returnResponse
+  }
+  
+  public func cancel() {
+    call.cancel()
+  }
+}
+
+/// Simple fake implementation of ClientCallServerStreamingBase that returns a previously-defined set of results.
+open class ClientCallServerStreamingTestStub<OutputType: Message>: ClientCallServerStreamingBase {
+  open class var method: String { fatalError("needs to be overridden") }
+  
+  var outputs: [OutputType] = []
+  
+  public func receive(completion:@escaping (OutputType?, ClientError?)->()) throws {
+    if let output = outputs.first {
+      outputs.removeFirst()
+      completion(output, nil)
+    } else {
+      completion(nil, .endOfStream)
+    }
+  }
+  
+  public func receive() throws -> OutputType {
+    if let output = outputs.first {
+      outputs.removeFirst()
+      return output
+    } else {
+      throw ClientError.endOfStream
+    }
+  }
+  
+  public func cancel() { }
+}