Adds GRPCPayload protocol to allow other types of payloads to be used with GRPC-swift

Author: mustii

.gitignore

@@ -3,6 +3,7 @@ project.xcworkspace
 xcuserdata
 DerivedData/
 .build
+.swiftpm
 build
 /protoc-gen-swift
 /protoc-gen-grpc-swift

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

@@ -149,3 +149,6 @@ extension Echo_EchoProvider {
   }
 }
 
+
+extension Echo_EchoRequest: GRPCProtobufPayload {}
+extension Echo_EchoResponse: GRPCProtobufPayload {}

Sources/Examples/HelloWorld/Model/helloworld.grpc.swift

@@ -84,3 +84,6 @@ extension Helloworld_GreeterProvider {
   }
 }
 
+
+extension Helloworld_HelloRequest: GRPCProtobufPayload {}
+extension Helloworld_HelloReply: GRPCProtobufPayload {}

Sources/Examples/RouteGuide/Model/route_guide.grpc.swift

@@ -149,3 +149,9 @@ extension Routeguide_RouteGuideProvider {
   }
 }
 
+
+extension Routeguide_Point: GRPCProtobufPayload {}
+extension Routeguide_Feature: GRPCProtobufPayload {}
+extension Routeguide_Rectangle: GRPCProtobufPayload {}
+extension Routeguide_RouteSummary: GRPCProtobufPayload {}
+extension Routeguide_RouteNote: GRPCProtobufPayload {}

Sources/GRPC/CallHandlers/BidirectionalStreamingCallHandler.swift

@@ -26,8 +26,8 @@ import Logging
 ///   they can fail the observer block future.
 /// - To close the call and send the status, complete `context.statusPromise`.
 public class BidirectionalStreamingCallHandler<
-  RequestMessage: Message,
-  ResponseMessage: Message
+  RequestMessage: GRPCPayload,
+  ResponseMessage: GRPCPayload
 >: _BaseCallHandler<RequestMessage, ResponseMessage> {
   public typealias Context = StreamingResponseCallContext<ResponseMessage>
   public typealias EventObserver = (StreamEvent<RequestMessage>) -> Void

Sources/GRPC/CallHandlers/ClientStreamingCallHandler.swift

@@ -34,8 +34,8 @@ enum ClientStreamingHandlerObserverState<Factory, Observer> {
 ///   they can fail the observer block future.
 /// - To close the call and send the response, complete `context.responsePromise`.
 public final class ClientStreamingCallHandler<
-  RequestMessage: Message,
-  ResponseMessage: Message
+  RequestMessage: GRPCPayload,
+  ResponseMessage: GRPCPayload
 >: _BaseCallHandler<RequestMessage, ResponseMessage> {
   public typealias Context = UnaryResponseCallContext<ResponseMessage>
   public typealias EventObserver = (StreamEvent<RequestMessage>) -> Void

Sources/GRPC/CallHandlers/ServerStreamingCallHandler.swift

@@ -24,8 +24,8 @@ import Logging
 /// - The observer block is implemented by the framework user and calls `context.sendResponse` as needed.
 /// - To close the call and send the status, complete the status future returned by the observer block.
 public final class ServerStreamingCallHandler<
-  RequestMessage: Message,
-  ResponseMessage: Message
+  RequestMessage: GRPCPayload,
+  ResponseMessage: GRPCPayload
 >: _BaseCallHandler<RequestMessage, ResponseMessage> {
   public typealias EventObserver = (RequestMessage) -> EventLoopFuture<GRPCStatus>
 

Sources/GRPC/CallHandlers/UnaryCallHandler.swift

@@ -25,8 +25,8 @@ import Logging
 /// - To return a response to the client, the framework user should complete that future
 ///   (similar to e.g. serving regular HTTP requests in frameworks such as Vapor).
 public final class UnaryCallHandler<
-  RequestMessage: Message,
-  ResponseMessage: Message
+  RequestMessage: GRPCPayload,
+  ResponseMessage: GRPCPayload
 >: _BaseCallHandler<RequestMessage, ResponseMessage> {
   public typealias EventObserver = (RequestMessage) -> EventLoopFuture<ResponseMessage>
   private var eventObserver: EventObserver?

Sources/GRPC/CallHandlers/_BaseCallHandler.swift

@@ -23,7 +23,7 @@ import Logging
 ///
 /// Calls through to `processMessage` for individual messages it receives, which needs to be implemented by subclasses.
 /// - Important: This is **NOT** part of the public API.
-public class _BaseCallHandler<RequestMessage: Message, ResponseMessage: Message>: GRPCCallHandler {
+public class _BaseCallHandler<RequestMessage: GRPCPayload, ResponseMessage: GRPCPayload>: GRPCCallHandler {
   public func makeGRPCServerCodec() -> ChannelHandler {
     return GRPCServerCodec<RequestMessage, ResponseMessage>()
   }

Sources/GRPC/ClientCalls/BaseClientCall.swift

@@ -48,7 +48,7 @@ import Logging
 ///
 /// This class also provides much of the framework user facing functionality via conformance to
 /// `ClientCall`.
-public class BaseClientCall<Request: Message, Response: Message>: ClientCall {
+public class BaseClientCall<Request: GRPCPayload, Response: GRPCPayload>: ClientCall {
   public typealias RequestMessage = Request
   public typealias ResponseMessage = Response
 

Sources/GRPC/ClientCalls/BidirectionalStreamingCall.swift

@@ -27,7 +27,7 @@ import Logging
 /// - `initialMetadata`: the initial metadata returned from the server,
 /// - `status`: the status of the gRPC call after it has ended,
 /// - `trailingMetadata`: any metadata returned from the server alongside the `status`.
-public final class BidirectionalStreamingCall<RequestMessage: Message, ResponseMessage: Message>
+public final class BidirectionalStreamingCall<RequestMessage: GRPCPayload, ResponseMessage: GRPCPayload>
   : BaseClientCall<RequestMessage, ResponseMessage>,
     StreamingRequestClientCall {
   private var messageQueue: EventLoopFuture<Void>

Sources/GRPC/ClientCalls/ClientCall.swift

@@ -23,9 +23,9 @@ import SwiftProtobuf
 /// Base protocol for a client call to a gRPC service.
 public protocol ClientCall {
   /// The type of the request message for the call.
-  associatedtype RequestMessage: Message
+  associatedtype RequestMessage: GRPCPayload
   /// The type of the response message for the call.
-  associatedtype ResponseMessage: Message
+  associatedtype ResponseMessage: GRPCPayload
 
   /// HTTP/2 stream that requests and responses are sent and received on.
   var subchannel: EventLoopFuture<Channel> { get }

Sources/GRPC/ClientCalls/ClientStreamingCall.swift

@@ -28,7 +28,7 @@ import Logging
 /// - `response`: the response from the call,
 /// - `status`: the status of the gRPC call after it has ended,
 /// - `trailingMetadata`: any metadata returned from the server alongside the `status`.
-public final class ClientStreamingCall<RequestMessage: Message, ResponseMessage: Message>
+public final class ClientStreamingCall<RequestMessage: GRPCPayload, ResponseMessage: GRPCPayload>
   : BaseClientCall<RequestMessage, ResponseMessage>,
     StreamingRequestClientCall,
     UnaryResponseClientCall {

Sources/GRPC/ClientCalls/ServerStreamingCall.swift

@@ -24,7 +24,7 @@ import Logging
 /// - `initialMetadata`: the initial metadata returned from the server,
 /// - `status`: the status of the gRPC call after it has ended,
 /// - `trailingMetadata`: any metadata returned from the server alongside the `status`.
-public final class ServerStreamingCall<RequestMessage: Message, ResponseMessage: Message>: BaseClientCall<RequestMessage, ResponseMessage> {
+public final class ServerStreamingCall<RequestMessage: GRPCPayload, ResponseMessage: GRPCPayload>: BaseClientCall<RequestMessage, ResponseMessage> {
   public init(
     connection: ClientConnection,
     path: String,

Sources/GRPC/ClientCalls/UnaryCall.swift

@@ -27,7 +27,7 @@ import Logging
 /// - `response`: the response from the unary call,
 /// - `status`: the status of the gRPC call after it has ended,
 /// - `trailingMetadata`: any metadata returned from the server alongside the `status`.
-public final class UnaryCall<RequestMessage: Message, ResponseMessage: Message>
+public final class UnaryCall<RequestMessage: GRPCPayload, ResponseMessage: GRPCPayload>
   : BaseClientCall<RequestMessage, ResponseMessage>,
     UnaryResponseClientCall {
   public let response: EventLoopFuture<ResponseMessage>

Sources/GRPC/GRPCClient.swift

@@ -26,7 +26,7 @@ public protocol GRPCClient {
 }
 
 extension GRPCClient {
-  public func makeUnaryCall<Request: Message, Response: Message>(
+  public func makeUnaryCall<Request: GRPCPayload, Response: GRPCPayload>(
     path: String,
     request: Request,
     callOptions: CallOptions? = nil,
@@ -40,7 +40,7 @@ extension GRPCClient {
       errorDelegate: self.connection.configuration.errorDelegate)
   }
 
-  public func makeServerStreamingCall<Request: Message, Response: Message>(
+  public func makeServerStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
     path: String,
     request: Request,
     callOptions: CallOptions? = nil,
@@ -56,7 +56,7 @@ extension GRPCClient {
       handler: handler)
   }
 
-  public func makeClientStreamingCall<Request: Message, Response: Message>(
+  public func makeClientStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
     path: String,
     callOptions: CallOptions? = nil,
     requestType: Request.Type = Request.self,
@@ -69,7 +69,7 @@ extension GRPCClient {
       errorDelegate: self.connection.configuration.errorDelegate)
   }
 
-  public func makeBidirectionalStreamingCall<Request: Message, Response: Message>(
+  public func makeBidirectionalStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
     path: String,
     callOptions: CallOptions? = nil,
     requestType: Request.Type = Request.self,

Sources/GRPC/GRPCClientResponseChannelHandler.swift

@@ -24,7 +24,7 @@ import Logging
 /// This includes holding promises for the initial metadata and status of the gRPC call. This handler
 /// is also responsible for error handling, via an error delegate and by appropriately failing the
 /// aforementioned promises.
-internal class GRPCClientResponseChannelHandler<ResponseMessage: Message>: ChannelInboundHandler {
+internal class GRPCClientResponseChannelHandler<ResponseMessage: GRPCPayload>: ChannelInboundHandler {
   public typealias InboundIn = _GRPCClientResponsePart<ResponseMessage>
   internal let logger: Logger
   internal var stopwatch: Stopwatch?
@@ -186,7 +186,7 @@ internal class GRPCClientResponseChannelHandler<ResponseMessage: Message>: Chann
 }
 
 /// A channel handler for client calls which receive a single response.
-final class GRPCClientUnaryResponseChannelHandler<ResponseMessage: Message>: GRPCClientResponseChannelHandler<ResponseMessage> {
+final class GRPCClientUnaryResponseChannelHandler<ResponseMessage: GRPCPayload>: GRPCClientResponseChannelHandler<ResponseMessage> {
   let responsePromise: EventLoopPromise<ResponseMessage>
 
   internal init(
@@ -236,7 +236,7 @@ final class GRPCClientUnaryResponseChannelHandler<ResponseMessage: Message>: GRP
 }
 
 /// A channel handler for client calls which receive a stream of responses.
-final class GRPCClientStreamingResponseChannelHandler<ResponseMessage: Message>: GRPCClientResponseChannelHandler<ResponseMessage> {
+final class GRPCClientStreamingResponseChannelHandler<ResponseMessage: GRPCPayload>: GRPCClientResponseChannelHandler<ResponseMessage> {
   typealias ResponseHandler = (ResponseMessage) -> Void
 
   let responseHandler: ResponseHandler

Sources/GRPC/GRPCClientStateMachine.swift

@@ -66,7 +66,7 @@ enum SendEndOfRequestStreamError: Error {
 /// A state machine for a single gRPC call from the perspective of a client.
 ///
 /// See: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
-struct GRPCClientStateMachine<Request: Message, Response: Message> {
+struct GRPCClientStateMachine<Request: GRPCPayload, Response: GRPCPayload> {
   /// The combined state of the request (client) and response (server) streams for an RPC call.
   ///
   /// The following states are not possible:

Sources/GRPC/GRPCPayload.swift

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2020, 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 NIO
+
+public protocol GRPCPayload {
+    init(serializedByteBuffer: inout NIO.ByteBuffer) throws
+    func serialize(into buffer: inout NIO.ByteBuffer) throws
+}

Sources/GRPC/GRPCProtobufPayload.swift

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2020, 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 NIO
+import SwiftProtobuf
+
+public protocol GRPCProtobufPayload: GRPCPayload, Message {}
+
+public extension GRPCProtobufPayload {
+    init(serializedByteBuffer: inout NIO.ByteBuffer) throws {
+        try self.init(serializedData: serializedByteBuffer.readData(length: serializedByteBuffer.readableBytes)!,
+                  extensions: nil,
+                  partial: false,
+                  options: BinaryDecodingOptions())
+    }
+    func serialize(into buffer: inout NIO.ByteBuffer) throws {
+        let data = try self.serializedData()
+        buffer.writeBytes(data)
+    }
+}

Sources/GRPC/GRPCServerCodec.swift

@@ -22,7 +22,7 @@ import NIOHTTP1
 /// Incoming gRPC package with a fixed message type.
 ///
 /// - Important: This is **NOT** part of the public API.
-public enum _GRPCServerRequestPart<RequestMessage: Message> {
+public enum _GRPCServerRequestPart<RequestMessage: GRPCPayload> {
   case head(HTTPRequestHead)
   case message(RequestMessage)
   case end
@@ -31,14 +31,14 @@ public enum _GRPCServerRequestPart<RequestMessage: Message> {
 /// Outgoing gRPC package with a fixed message type.
 ///
 /// - Important: This is **NOT** part of the public API.
-public enum _GRPCServerResponsePart<ResponseMessage: Message> {
+public enum _GRPCServerResponsePart<ResponseMessage: GRPCPayload> {
   case headers(HTTPHeaders)
   case message(ResponseMessage)
   case statusAndTrailers(GRPCStatus, HTTPHeaders)
 }
 
 /// A simple channel handler that translates raw gRPC packets into decoded protobuf messages, and vice versa.
-internal final class GRPCServerCodec<RequestMessage: Message, ResponseMessage: Message> {}
+internal final class GRPCServerCodec<RequestMessage: GRPCPayload, ResponseMessage: GRPCPayload> {}
 
 extension GRPCServerCodec: ChannelInboundHandler {
   typealias InboundIn = _RawGRPCServerRequestPart
@@ -50,9 +50,8 @@ extension GRPCServerCodec: ChannelInboundHandler {
       context.fireChannelRead(self.wrapInboundOut(.head(requestHead)))
 
     case .message(var message):
-      let messageAsData = message.readData(length: message.readableBytes)!
       do {
-        context.fireChannelRead(self.wrapInboundOut(.message(try RequestMessage(serializedData: messageAsData))))
+        context.fireChannelRead(self.wrapInboundOut(.message(try RequestMessage(serializedByteBuffer: &message))))
       } catch {
         context.fireErrorCaught(GRPCError.DeserializationFailure().captureContext())
       }
@@ -75,8 +74,9 @@ extension GRPCServerCodec: ChannelOutboundHandler {
 
     case .message(let message):
       do {
-        let messageData = try message.serializedData()
-        context.write(self.wrapOutboundOut(.message(messageData)), promise: promise)
+        var buffer = ByteBufferAllocator().buffer(capacity: 0)
+        try message.serialize(into: &buffer)
+        context.write(self.wrapOutboundOut(.message(buffer.readData(length: buffer.readableBytes)!)), promise: promise)
       } catch {
         let error = GRPCError.SerializationFailure().captureContext()
         promise?.fail(error)

Sources/GRPC/ReadWriteStates.swift

@@ -54,7 +54,7 @@ enum WriteState {
   /// - Parameter allocator: An allocator to provide a `ByteBuffer` into which the message will be
   ///     written.
   mutating func write(
-    _ message: Message,
+    _ message: GRPCPayload,
     disableCompression: Bool,
     allocator: ByteBufferAllocator
   ) -> Result<ByteBuffer, MessageWriteError> {
@@ -63,13 +63,19 @@ enum WriteState {
       return .failure(.cardinalityViolation)
 
     case let .writing(writeArity, contentType, writer):
-      guard let data = try? message.serializedData() else {
+      // Zero is fine: the writer will allocate the correct amount of space.
+      var buffer = allocator.buffer(capacity: 0)
+      
+      do {
+        try message.serialize(into: &buffer)
+      } catch {
         self = .notWriting
         return .failure(.serializationFailed)
       }
-
+      
       // Zero is fine: the writer will allocate the correct amount of space.
-      var buffer = allocator.buffer(capacity: 0)
+      let data = buffer.readData(length: buffer.readableBytes)!
+      buffer.clear()
       do {
         try writer.write(data, into: &buffer, disableCompression: disableCompression)
       } catch {
@@ -115,7 +121,7 @@ enum ReadState {
   /// a message has been produced then subsequent calls will result in an error.
   ///
   /// - Parameter buffer: The buffer to read from.
-  mutating func readMessages<MessageType: Message>(
+  mutating func readMessages<MessageType: GRPCPayload>(
     _ buffer: inout ByteBuffer,
     as: MessageType.Type = MessageType.self
   ) -> Result<[MessageType], MessageReadError> {
@@ -130,8 +136,7 @@ enum ReadState {
       do {
         while var serializedBytes = try? reader.nextMessage() {
           // Force unwrapping is okay here: we will always be able to read `readableBytes`.
-          let serializedData = serializedBytes.readData(length: serializedBytes.readableBytes)!
-          messages.append(try MessageType(serializedData: serializedData))
+          messages.append(try MessageType(serializedByteBuffer: &serializedBytes))
         }
       } catch {
         self = .notReading

Sources/GRPC/ServerCallContexts/StreamingResponseCallContext.swift

@@ -25,7 +25,7 @@ import Logging
 /// - If `statusPromise` is failed and the error is of type `GRPCStatusTransformable`,
 ///   the result of `error.asGRPCStatus()` will be returned to the client.
 /// - If `error.asGRPCStatus()` is not available, `GRPCStatus.processingError` is returned to the client.
-open class StreamingResponseCallContext<ResponseMessage: Message>: ServerCallContextBase {
+open class StreamingResponseCallContext<ResponseMessage: GRPCPayload>: ServerCallContextBase {
   typealias WrappedResponse = _GRPCServerResponsePart<ResponseMessage>
 
   public let statusPromise: EventLoopPromise<GRPCStatus>
@@ -41,7 +41,7 @@ open class StreamingResponseCallContext<ResponseMessage: Message>: ServerCallCon
 }
 
 /// Concrete implementation of `StreamingResponseCallContext` used by our generated code.
-open class StreamingResponseCallContextImpl<ResponseMessage: Message>: StreamingResponseCallContext<ResponseMessage> {
+open class StreamingResponseCallContextImpl<ResponseMessage: GRPCPayload>: StreamingResponseCallContext<ResponseMessage> {
   public let channel: Channel
 
   /// - Parameters:
@@ -80,7 +80,7 @@ open class StreamingResponseCallContextImpl<ResponseMessage: Message>: Streaming
 /// Concrete implementation of `StreamingResponseCallContext` used for testing.
 ///
 /// Simply records all sent messages.
-open class StreamingResponseCallContextTestStub<ResponseMessage: Message>: StreamingResponseCallContext<ResponseMessage> {
+open class StreamingResponseCallContextTestStub<ResponseMessage: GRPCPayload>: StreamingResponseCallContext<ResponseMessage> {
   open var recordedResponses: [ResponseMessage] = []
 
   open override func sendResponse(_ message: ResponseMessage) -> EventLoopFuture<Void> {