Adds the ability to use different payloads (#710)

Motivation:

gRPC should be agnostic to the type of payload that it sends to the peer to allow for other message types to be used (e.g. Flatbuffers).

Modifications:

- Make the generated code generic over `GRPCPayload`
- Provide a default implementation of `GRPCPayload` for `SwiftProtobuf.Message` in `GRPCProtobufPayload`
- Update generated code such that request and response types conform to `GRPCProtobufPayload`
- Shuffle around some of the server code to minimise payload copies

Result:

gRPC can send/receive payloads which conform to `GRPCPayload` instead of `SwiftProtobuf.Message`

Author: mustiikhalil

.gitignore

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

Makefile

@@ -31,7 +31,7 @@ plugins: ${PROTOC_GEN_SWIFT} ${PROTOC_GEN_GRPC_SWIFT}
 ${PROTOC_GEN_SWIFT}:
 	${SWIFT_BUILD_RELEASE} --product protoc-gen-swift
 
-${PROTOC_GEN_GRPC_SWIFT}:
+${PROTOC_GEN_GRPC_SWIFT}: Sources/protoc-gen-grpc-swift/*.swift
 	${SWIFT_BUILD_RELEASE} --product protoc-gen-grpc-swift
 
 interop-test-runner:

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

@@ -149,3 +149,7 @@ extension Echo_EchoProvider {
   }
 }
 
+
+/// Provides conformance to `GRPCPayload` for the request and response messages
+extension Echo_EchoRequest: GRPCProtobufPayload {}
+extension Echo_EchoResponse: GRPCProtobufPayload {}

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

@@ -84,3 +84,7 @@ extension Helloworld_GreeterProvider {
   }
 }
 
+
+/// Provides conformance to `GRPCPayload` for the request and response messages
+extension Helloworld_HelloRequest: GRPCProtobufPayload {}
+extension Helloworld_HelloReply: GRPCProtobufPayload {}

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

@@ -149,3 +149,10 @@ extension Routeguide_RouteGuideProvider {
   }
 }
 
+
+/// Provides conformance to `GRPCPayload` for the request and response messages
+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,11 +26,11 @@ 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
->: _BaseCallHandler<RequestMessage, ResponseMessage> {
-  public typealias Context = StreamingResponseCallContext<ResponseMessage>
-  public typealias EventObserver = (StreamEvent<RequestMessage>) -> Void
+  RequestPayload: GRPCPayload,
+  ResponsePayload: GRPCPayload
+>: _BaseCallHandler<RequestPayload, ResponsePayload> {
+  public typealias Context = StreamingResponseCallContext<ResponsePayload>
+  public typealias EventObserver = (StreamEvent<RequestPayload>) -> Void
   public typealias EventObserverFactory = (Context) -> EventLoopFuture<EventObserver>
 
   private var observerState: ClientStreamingHandlerObserverState<EventObserverFactory, EventObserver> {
@@ -44,7 +44,7 @@ public class BidirectionalStreamingCallHandler<
   // If authentication fails, they can simply fail the observer future, which causes the call to be terminated.
   public init(
     callHandlerContext: CallHandlerContext,
-    eventObserverFactory: @escaping (StreamingResponseCallContext<ResponseMessage>) -> EventLoopFuture<EventObserver>
+    eventObserverFactory: @escaping (StreamingResponseCallContext<ResponsePayload>) -> EventLoopFuture<EventObserver>
   ) {
     // Delay the creation of the event observer until `handlerAdded(context:)`, otherwise it is
     // possible for the service to write into the pipeline (by fulfilling the status promise
@@ -53,7 +53,7 @@ public class BidirectionalStreamingCallHandler<
 
     super.init(callHandlerContext: callHandlerContext)
 
-    let context = StreamingResponseCallContextImpl<ResponseMessage>(
+    let context = StreamingResponseCallContextImpl<ResponsePayload>(
       channel: self.callHandlerContext.channel,
       request: self.callHandlerContext.request,
       errorDelegate: self.callHandlerContext.errorDelegate,
@@ -81,12 +81,12 @@ public class BidirectionalStreamingCallHandler<
 
     // Terminate the call if the future providing an observer fails.
     // This is being done _after_ we have been added as a handler to ensure that the `GRPCServerCodec` required to
-    // translate our outgoing `GRPCServerResponsePart<ResponseMessage>` message is already present on the channel.
+    // translate our outgoing `GRPCServerResponsePart<ResponsePayload>` message is already present on the channel.
     // Otherwise, our `OutboundOut` type would not match the `OutboundIn` type of the next handler on the channel.
     eventObserver.cascadeFailure(to: callContext.statusPromise)
   }
 
-  internal override func processMessage(_ message: RequestMessage) {
+  internal override func processMessage(_ message: RequestPayload) {
     guard case .created(let eventObserver) = self.observerState else {
       self.logger.warning("expecting observerState to be .created but was \(self.observerState), ignoring message \(message)")
       return

Sources/GRPC/CallHandlers/ClientStreamingCallHandler.swift

@@ -34,19 +34,19 @@ 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
->: _BaseCallHandler<RequestMessage, ResponseMessage> {
-  public typealias Context = UnaryResponseCallContext<ResponseMessage>
-  public typealias EventObserver = (StreamEvent<RequestMessage>) -> Void
+  RequestPayload: GRPCPayload,
+  ResponsePayload: GRPCPayload
+>: _BaseCallHandler<RequestPayload, ResponsePayload> {
+  public typealias Context = UnaryResponseCallContext<ResponsePayload>
+  public typealias EventObserver = (StreamEvent<RequestPayload>) -> Void
   public typealias EventObserverFactory = (Context) -> EventLoopFuture<EventObserver>
 
   private var observerState: ClientStreamingHandlerObserverState<EventObserverFactory, EventObserver> {
     willSet(newState) {
       self.logger.debug("observerState changed from \(self.observerState) to \(newState)")
     }
   }
-  private var callContext: UnaryResponseCallContext<ResponseMessage>?
+  private var callContext: UnaryResponseCallContext<ResponsePayload>?
 
   // We ask for a future of type `EventObserver` to allow the framework user to e.g. asynchronously authenticate a call.
   // If authentication fails, they can simply fail the observer future, which causes the call to be terminated.
@@ -58,7 +58,7 @@ public final class ClientStreamingCallHandler<
 
     super.init(callHandlerContext: callHandlerContext)
 
-    let callContext = UnaryResponseCallContextImpl<ResponseMessage>(
+    let callContext = UnaryResponseCallContextImpl<ResponsePayload>(
       channel: self.callHandlerContext.channel,
       request: self.callHandlerContext.request,
       errorDelegate: self.callHandlerContext.errorDelegate,
@@ -86,12 +86,12 @@ public final class ClientStreamingCallHandler<
 
     // Terminate the call if the future providing an observer fails.
     // This is being done _after_ we have been added as a handler to ensure that the `GRPCServerCodec` required to
-    // translate our outgoing `GRPCServerResponsePart<ResponseMessage>` message is already present on the channel.
+    // translate our outgoing `GRPCServerResponsePart<ResponsePayload>` message is already present on the channel.
     // Otherwise, our `OutboundOut` type would not match the `OutboundIn` type of the next handler on the channel.
     eventObserver.cascadeFailure(to: callContext.responsePromise)
   }
 
-  internal override func processMessage(_ message: RequestMessage) {
+  internal override func processMessage(_ message: RequestPayload) {
     guard case .created(let eventObserver) = self.observerState else {
       self.logger.warning("expecting observerState to be .created but was \(self.observerState), ignoring message \(message)")
       return

Sources/GRPC/CallHandlers/ServerStreamingCallHandler.swift

@@ -24,20 +24,20 @@ 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
->: _BaseCallHandler<RequestMessage, ResponseMessage> {
-  public typealias EventObserver = (RequestMessage) -> EventLoopFuture<GRPCStatus>
+  RequestPayload: GRPCPayload,
+  ResponsePayload: GRPCPayload
+>: _BaseCallHandler<RequestPayload, ResponsePayload> {
+  public typealias EventObserver = (RequestPayload) -> EventLoopFuture<GRPCStatus>
 
   private var eventObserver: EventObserver?
-  private var callContext: StreamingResponseCallContext<ResponseMessage>?
+  private var callContext: StreamingResponseCallContext<ResponsePayload>?
 
   public init(
     callHandlerContext: CallHandlerContext,
-    eventObserverFactory: (StreamingResponseCallContext<ResponseMessage>) -> EventObserver
+    eventObserverFactory: (StreamingResponseCallContext<ResponsePayload>) -> EventObserver
   ) {
     super.init(callHandlerContext: callHandlerContext)
-    let callContext = StreamingResponseCallContextImpl<ResponseMessage>(
+    let callContext = StreamingResponseCallContextImpl<ResponsePayload>(
       channel: self.callHandlerContext.channel,
       request: self.callHandlerContext.request,
       errorDelegate: self.callHandlerContext.errorDelegate,
@@ -53,7 +53,7 @@ public final class ServerStreamingCallHandler<
     }
   }
 
-  override internal func processMessage(_ message: RequestMessage) throws {
+  override internal func processMessage(_ message: RequestPayload) throws {
     guard let eventObserver = self.eventObserver,
       let callContext = self.callContext else {
         self.logger.error("processMessage(_:) called before the call started or after the call completed")

Sources/GRPC/CallHandlers/UnaryCallHandler.swift

@@ -25,19 +25,19 @@ 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
->: _BaseCallHandler<RequestMessage, ResponseMessage> {
-  public typealias EventObserver = (RequestMessage) -> EventLoopFuture<ResponseMessage>
+  RequestPayload: GRPCPayload,
+  ResponsePayload: GRPCPayload
+>: _BaseCallHandler<RequestPayload, ResponsePayload> {
+  public typealias EventObserver = (RequestPayload) -> EventLoopFuture<ResponsePayload>
   private var eventObserver: EventObserver?
-  private var callContext: UnaryResponseCallContext<ResponseMessage>?
+  private var callContext: UnaryResponseCallContext<ResponsePayload>?
 
   public init(
     callHandlerContext: CallHandlerContext,
-    eventObserverFactory: (UnaryResponseCallContext<ResponseMessage>) -> EventObserver
+    eventObserverFactory: (UnaryResponseCallContext<ResponsePayload>) -> EventObserver
   ) {
     super.init(callHandlerContext: callHandlerContext)
-    let callContext = UnaryResponseCallContextImpl<ResponseMessage>(
+    let callContext = UnaryResponseCallContextImpl<ResponsePayload>(
       channel: self.callHandlerContext.channel,
       request: self.callHandlerContext.request,
       errorDelegate: self.callHandlerContext.errorDelegate,
@@ -53,7 +53,7 @@ public final class UnaryCallHandler<
     }
   }
 
-  internal override func processMessage(_ message: RequestMessage) throws {
+  internal override func processMessage(_ message: RequestPayload) throws {
     guard let eventObserver = self.eventObserver,
       let context = self.callContext else {
       self.logger.error("processMessage(_:) called before the call started or after the call completed")

Sources/GRPC/CallHandlers/_BaseCallHandler.swift

@@ -23,15 +23,15 @@ 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<RequestPayload: GRPCPayload, ResponsePayload: GRPCPayload>: GRPCCallHandler {
   public func makeGRPCServerCodec() -> ChannelHandler {
-    return GRPCServerCodec<RequestMessage, ResponseMessage>()
+    return GRPCServerCodec<RequestPayload, ResponsePayload>()
   }
 
   /// Called whenever a message has been received.
   ///
   /// Overridden by subclasses.
-  internal func processMessage(_ message: RequestMessage) throws {
+  internal func processMessage(_ message: RequestPayload) throws {
     fatalError("needs to be overridden")
   }
 
@@ -71,7 +71,7 @@ public class _BaseCallHandler<RequestMessage: Message, ResponseMessage: Message>
 }
 
 extension _BaseCallHandler: ChannelInboundHandler {
-  public typealias InboundIn = _GRPCServerRequestPart<RequestMessage>
+  public typealias InboundIn = _GRPCServerRequestPart<RequestPayload>
 
   /// Passes errors to the user-provided `errorHandler`. After an error has been received an
   /// appropriate status is written. Errors which don't conform to `GRPCStatusTransformable`
@@ -120,8 +120,8 @@ extension _BaseCallHandler: ChannelInboundHandler {
 }
 
 extension _BaseCallHandler: ChannelOutboundHandler {
-  public typealias OutboundIn = _GRPCServerResponsePart<ResponseMessage>
-  public typealias OutboundOut = _GRPCServerResponsePart<ResponseMessage>
+  public typealias OutboundIn = _GRPCServerResponsePart<ResponsePayload>
+  public typealias OutboundOut = _GRPCServerResponsePart<ResponsePayload>
 
   public func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
     guard self.serverCanWrite else {

Sources/GRPC/ClientCalls/BaseClientCall.swift

@@ -48,9 +48,9 @@ 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 typealias RequestMessage = Request
-  public typealias ResponseMessage = Response
+public class BaseClientCall<Request: GRPCPayload, Response: GRPCPayload>: ClientCall {
+  public typealias RequestPayload = Request
+  public typealias ResponsePayload = Response
 
   internal let logger: Logger
 

Sources/GRPC/ClientCalls/BidirectionalStreamingCall.swift

@@ -27,8 +27,8 @@ 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>
-  : BaseClientCall<RequestMessage, ResponseMessage>,
+public final class BidirectionalStreamingCall<RequestPayload: GRPCPayload, ResponsePayload: GRPCPayload>
+  : BaseClientCall<RequestPayload, ResponsePayload>,
     StreamingRequestClientCall {
   private var messageQueue: EventLoopFuture<Void>
 
@@ -37,7 +37,7 @@ public final class BidirectionalStreamingCall<RequestMessage: Message, ResponseM
     path: String,
     callOptions: CallOptions,
     errorDelegate: ClientErrorDelegate?,
-    handler: @escaping (ResponseMessage) -> Void
+    handler: @escaping (ResponsePayload) -> Void
   ) {
     self.messageQueue = connection.channel.eventLoop.makeSucceededFuture(())
     let requestID = callOptions.requestIDProvider.requestID()
@@ -64,7 +64,7 @@ public final class BidirectionalStreamingCall<RequestMessage: Message, ResponseM
       options: callOptions
     )
 
-    let requestHandler = _StreamingRequestChannelHandler<RequestMessage>(requestHead: requestHead)
+    let requestHandler = _StreamingRequestChannelHandler<RequestPayload>(requestHead: requestHead)
 
     super.init(
       eventLoop: connection.eventLoop,