Ensure that errors in client calls are always provided to the user.

The expected behavior is now for the client call's completion handler to be called when the whole call is completed. You can check that completion handler's `statusCode` for the call's result. The intermediate send/receive call completion handlers will also get called with an error, but that error's status code is most likely to be `.unknown` (because send/receive calls don't receive the final status from the server).
In addition, `CallResult` now also includes the operation group's success flag, but that flag is much less important now.

This includes:
- Make most completion handlers non-throwing (their errors were for the most part discarded, anyway)
- Extract Receive-Streaming into a common protocol
- Changing many completion handler interfaces to be called with a `CallResult` instead of `Data?` and `ResultOrRPCError<ReceivedType?>` instead of `(ReceivedType, Error)`
- Not throwing on end of stream, but passing `nil` to the completion handler instead
- Making `ServerSessionUnary` always return an error when one of its calls fail
- Adding test to ensure the proper error behavior.

Author: MrMage

.gitignore

@@ -5,11 +5,6 @@ xcuserdata
 protoc-gen-swift
 protoc-gen-swiftgrpc
 third_party/**
-Plugin/Packages/**
-Plugin/Sources/protoc-gen-swiftgrpc/templates.swift
-Plugin/protoc-*
-Plugin/swiftgrpc.log
-Plugin/echo.*.swift
 Echo
 test.out
 echo.pid

Sources/Examples/Echo/EchoProvider.swift

@@ -45,10 +45,9 @@ class EchoProvider: Echo_EchoProvider {
     var parts: [String] = []
     while true {
       do {
-        let request = try session.receive()
+        guard let request = try session.receive()
+          else { break }  // End of stream
         parts.append(request.text)
-      } catch ServerError.endOfStream {
-        break
       } catch (let error) {
         print("\(error)")
       }
@@ -63,7 +62,8 @@ class EchoProvider: Echo_EchoProvider {
     var count = 0
     while true {
       do {
-        let request = try session.receive()
+        guard let request = try session.receive()
+          else { break }  // End of stream
         var response = Echo_EchoResponse()
         response.text = "Swift echo update (\(count)): \(request.text)"
         count += 1
@@ -72,8 +72,6 @@ class EchoProvider: Echo_EchoProvider {
             print("update error: \(error)")
           }
         }
-      } catch ServerError.endOfStream {
-        break
       } catch (let error) {
         print("\(error)")
         break

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

@@ -33,9 +33,9 @@ fileprivate final class Echo_EchoGetCallBase: ClientCallUnaryBase<Echo_EchoReque
 
 internal protocol Echo_EchoExpandCall: ClientCallServerStreaming {
   /// Call this to wait for a result. Blocking.
-  func receive() throws -> Echo_EchoResponse
+  func receive() throws -> Echo_EchoResponse?
   /// Call this to wait for a result. Nonblocking.
-  func receive(completion: @escaping (Echo_EchoResponse?, ClientError?) -> Void) throws
+  func receive(completion: @escaping (ResultOrRPCError<Echo_EchoResponse?>) -> Void) throws
 }
 
 fileprivate final class Echo_EchoExpandCallBase: ClientCallServerStreamingBase<Echo_EchoRequest, Echo_EchoResponse>, Echo_EchoExpandCall {
@@ -53,7 +53,7 @@ internal protocol Echo_EchoCollectCall: ClientCallClientStreaming {
   /// Call this to close the connection and wait for a response. Blocking.
   func closeAndReceive() throws -> Echo_EchoResponse
   /// Call this to close the connection and wait for a response. Nonblocking.
-  func closeAndReceive(completion: @escaping (Echo_EchoResponse?, ClientError?) -> Void) throws
+  func closeAndReceive(completion: @escaping (ResultOrRPCError<Echo_EchoResponse>) -> Void) throws
 }
 
 fileprivate final class Echo_EchoCollectCallBase: ClientCallClientStreamingBase<Echo_EchoRequest, Echo_EchoResponse>, Echo_EchoCollectCall {
@@ -68,9 +68,9 @@ class Echo_EchoCollectCallTestStub: ClientCallClientStreamingTestStub<Echo_EchoR
 
 internal protocol Echo_EchoUpdateCall: ClientCallBidirectionalStreaming {
   /// Call this to wait for a result. Blocking.
-  func receive() throws -> Echo_EchoResponse
+  func receive() throws -> Echo_EchoResponse?
   /// Call this to wait for a result. Nonblocking.
-  func receive(completion: @escaping (Echo_EchoResponse?, ClientError?) -> Void) throws
+  func receive(completion: @escaping (ResultOrRPCError<Echo_EchoResponse?>) -> Void) throws
 
   /// Call this to send each message in the request stream.
   func send(_ message: Echo_EchoRequest, completion: @escaping (Error?) -> Void) throws
@@ -210,8 +210,10 @@ fileprivate final class Echo_EchoExpandSessionBase: ServerSessionServerStreaming
 class Echo_EchoExpandSessionTestStub: ServerSessionServerStreamingTestStub<Echo_EchoResponse>, Echo_EchoExpandSession {}
 
 internal protocol Echo_EchoCollectSession: ServerSessionClientStreaming {
-  /// Receive a message. Blocks until a message is received or the client closes the connection.
-  func receive() throws -> Echo_EchoRequest
+  /// Call this to wait for a result. Blocking.
+  func receive() throws -> Echo_EchoRequest?
+  /// Call this to wait for a result. Nonblocking.
+  func receive(completion: @escaping (ResultOrRPCError<Echo_EchoRequest?>) -> Void) throws
 
   /// Send a response and close the connection.
   func sendAndClose(_ response: Echo_EchoResponse) throws
@@ -222,8 +224,10 @@ fileprivate final class Echo_EchoCollectSessionBase: ServerSessionClientStreamin
 class Echo_EchoCollectSessionTestStub: ServerSessionClientStreamingTestStub<Echo_EchoRequest, Echo_EchoResponse>, Echo_EchoCollectSession {}
 
 internal protocol Echo_EchoUpdateSession: ServerSessionBidirectionalStreaming {
-  /// Receive a message. Blocks until a message is received or the client closes the connection.
-  func receive() throws -> Echo_EchoRequest
+  /// Call this to wait for a result. Blocking.
+  func receive() throws -> Echo_EchoRequest?
+  /// Call this to wait for a result. Nonblocking.
+  func receive(completion: @escaping (ResultOrRPCError<Echo_EchoRequest?>) -> Void) throws
 
   /// Send a message. Nonblocking.
   func send(_ response: Echo_EchoResponse, completion: ((Error?) -> Void)?) throws

Sources/Examples/Echo/main.swift

@@ -109,16 +109,13 @@ Group {
       callResult = result
       sem.signal()
     }
-    var running = true
-    while running {
-      do {
-        let responseMessage = try expandCall.receive()
-        print("expand received: \(responseMessage.text)")
-      } catch ClientError.endOfStream {
-        running = false
-      }
+    while true {
+      guard let responseMessage = try expandCall.receive()
+        else { break }  // End of stream
+      print("expand received: \(responseMessage.text)")
     }
     _ = sem.wait()
+
     if let statusCode = callResult?.statusCode {
       print("expand completed with code \(statusCode)")
     }
@@ -150,6 +147,7 @@ Group {
     let responseMessage = try collectCall.closeAndReceive()
     print("collect received: \(responseMessage.text)")
     _ = sem.wait()
+
     if let statusCode = callResult?.statusCode {
       print("collect completed with code \(statusCode)")
     }
@@ -182,17 +180,10 @@ Group {
     try updateCall.closeSend()
 
     while true {
-      do {
-        let responseMessage = try updateCall.receive()
-        print("update received: \(responseMessage.text)")
-      } catch ClientError.endOfStream {
-        break
-      } catch (let error) {
-        print("update receive error: \(error)")
-        break
-      }
+      guard let responseMessage = try updateCall.receive()
+        else { break }  // End of stream
+      print("update received: \(responseMessage.text)")
     }
-
     _ = sem.wait()
 
     if let statusCode = callResult?.statusCode {

Sources/SwiftGRPC/Core/Call.swift

@@ -30,113 +30,6 @@ public enum CallWarning: Error {
   case blocked
 }
 
-public enum CallError: Error {
-  case ok
-  case unknown
-  case notOnServer
-  case notOnClient
-  case alreadyAccepted
-  case alreadyInvoked
-  case notInvoked
-  case alreadyFinished
-  case tooManyOperations
-  case invalidFlags
-  case invalidMetadata
-  case invalidMessage
-  case notServerCompletionQueue
-  case batchTooBig
-  case payloadTypeMismatch
-
-  static func callError(grpcCallError error: grpc_call_error) -> CallError {
-    switch error {
-    case GRPC_CALL_OK:
-      return .ok
-    case GRPC_CALL_ERROR:
-      return .unknown
-    case GRPC_CALL_ERROR_NOT_ON_SERVER:
-      return .notOnServer
-    case GRPC_CALL_ERROR_NOT_ON_CLIENT:
-      return .notOnClient
-    case GRPC_CALL_ERROR_ALREADY_ACCEPTED:
-      return .alreadyAccepted
-    case GRPC_CALL_ERROR_ALREADY_INVOKED:
-      return .alreadyInvoked
-    case GRPC_CALL_ERROR_NOT_INVOKED:
-      return .notInvoked
-    case GRPC_CALL_ERROR_ALREADY_FINISHED:
-      return .alreadyFinished
-    case GRPC_CALL_ERROR_TOO_MANY_OPERATIONS:
-      return .tooManyOperations
-    case GRPC_CALL_ERROR_INVALID_FLAGS:
-      return .invalidFlags
-    case GRPC_CALL_ERROR_INVALID_METADATA:
-      return .invalidMetadata
-    case GRPC_CALL_ERROR_INVALID_MESSAGE:
-      return .invalidMessage
-    case GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE:
-      return .notServerCompletionQueue
-    case GRPC_CALL_ERROR_BATCH_TOO_BIG:
-      return .batchTooBig
-    case GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH:
-      return .payloadTypeMismatch
-    default:
-      return .unknown
-    }
-  }
-}
-
-public struct CallResult: CustomStringConvertible {
-  public let statusCode: StatusCode
-  public let statusMessage: String?
-  public let resultData: Data?
-  public let initialMetadata: Metadata?
-  public let trailingMetadata: Metadata?
-
-  fileprivate init(_ op: OperationGroup) {
-    if op.success {
-      if let statusCodeRawValue = op.receivedStatusCode() {
-        if let statusCode = StatusCode(rawValue: statusCodeRawValue) {
-          self.statusCode = statusCode
-        } else {
-          statusCode = .unknown
-        }
-      } else {
-        statusCode = .ok
-      }
-      statusMessage = op.receivedStatusMessage()
-      resultData = op.receivedMessage()?.data()
-      initialMetadata = op.receivedInitialMetadata()
-      trailingMetadata = op.receivedTrailingMetadata()
-    } else {
-      statusCode = .unknown
-      statusMessage = nil
-      resultData = nil
-      initialMetadata = nil
-      trailingMetadata = nil
-    }
-  }
-
-  public var description: String {
-    var result = "status \(statusCode)"
-    if let statusMessage = self.statusMessage {
-      result += ": " + statusMessage
-    }
-    if let resultData = self.resultData {
-      result += "\n"
-      result += resultData.description
-    }
-    if let initialMetadata = self.initialMetadata {
-      result += "\n"
-      result += initialMetadata.description
-    }
-    if let trailingMetadata = self.trailingMetadata {
-      result += "\n"
-      result += trailingMetadata.description
-    }
-    return result
-  }
-}
-
 /// A gRPC API call
 public class Call {
   /// Shared mutex for synchronizing calls to cgrpc_call_perform()
@@ -240,11 +133,18 @@ public class Call {
         .receiveStatusOnClient,
       ]
     case .clientStreaming, .bidiStreaming:
-      operations = [
-        .sendInitialMetadata(metadata.copy()),
-        .receiveInitialMetadata,
-        .receiveStatusOnClient,
-      ]
+      try perform(OperationGroup(call: self,
+                                 operations: [
+                                  .sendInitialMetadata(metadata.copy()),
+                                  .receiveInitialMetadata
+                                  ],
+                                 completion: nil))
+      try perform(OperationGroup(call: self,
+                                 operations: [.receiveStatusOnClient],
+                                 completion: completion != nil
+                                  ? { op in completion?(CallResult(op)) }
+                                  : nil))
+      return
     }
     try perform(OperationGroup(call: self,
                                operations: operations,
@@ -304,27 +204,17 @@ public class Call {
 
   // Receive a message over a streaming connection.
   /// - Throws: `CallError` if fails to call.
-  public func closeAndReceiveMessage(completion: @escaping (Data?) throws -> Void) throws {
+  public func closeAndReceiveMessage(completion: @escaping (CallResult) -> Void) throws {
     try perform(OperationGroup(call: self, operations: [.sendCloseFromClient, .receiveMessage]) { operationGroup in
-      if operationGroup.success {
-        if let messageBuffer = operationGroup.receivedMessage() {
-          try completion(messageBuffer.data())
-        } else {
-          try completion(nil) // an empty response signals the end of a connection
-        }
-      }
+      completion(CallResult(operationGroup))
     })
   }
 
   // Receive a message over a streaming connection.
   /// - Throws: `CallError` if fails to call.
-  public func receiveMessage(completion: @escaping (Data?) throws -> Void) throws {
+  public func receiveMessage(completion: @escaping (CallResult) -> Void) throws {
     try perform(OperationGroup(call: self, operations: [.receiveMessage]) { operationGroup in
-      if operationGroup.success {
-        try completion(operationGroup.receivedMessage()?.data())
-      } else {
-        try completion(nil)
-      }
+      completion(CallResult(operationGroup))
     })
   }