First implementation of gRPC based on SwiftNIO

.travis.yml

@@ -34,6 +34,8 @@ sudo: false
 
 addons:
   apt:
+    sources:
+      - sourceline: 'ppa:ondrej/apache2'  # for libnghttp2-dev
     packages:
       - clang-3.8 
       - lldb-3.8 
@@ -46,6 +48,10 @@ addons:
       - uuid-dev
       - curl
       - unzip
+      - libnghttp2-dev
+
+before_install:
+  - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then brew install nghttp2; fi
 
 install: ./.travis-install.sh
 

Makefile

@@ -45,6 +45,11 @@ test-plugin:
 	protoc Sources/Examples/Echo/echo.proto --proto_path=Sources/Examples/Echo --plugin=.build/debug/protoc-gen-swift --plugin=.build/debug/protoc-gen-swiftgrpc --swiftgrpc_out=/tmp --swiftgrpc_opt=TestStubs=true
 	diff -u /tmp/echo.grpc.swift Sources/Examples/Echo/Generated/echo.grpc.swift
 
+test-plugin-nio:
+	swift build $(CFLAGS) --product protoc-gen-swiftgrpc
+	protoc Sources/Examples/Echo/echo.proto --proto_path=Sources/Examples/Echo --plugin=.build/debug/protoc-gen-swift --plugin=.build/debug/protoc-gen-swiftgrpc --swiftgrpc_out=/tmp --swiftgrpc_opt=Client=false,NIO=true
+		diff -u /tmp/echo.grpc.swift Tests/SwiftGRPCNIOTests/echo_nio.grpc.swift
+
 xcodebuild: project
 		xcodebuild -project SwiftGRPC.xcodeproj -configuration "Debug" -parallelizeTargets -target SwiftGRPC -target Echo -target Simple -target protoc-gen-swiftgrpc build
 

Package.swift

@@ -19,7 +19,10 @@ import PackageDescription
 var packageDependencies: [Package.Dependency] = [
   .package(url: "https://github.com/apple/swift-protobuf.git", .upToNextMinor(from: "1.1.1")),
   .package(url: "https://github.com/kylef/Commander.git", from: "0.8.0"),
-  .package(url: "https://github.com/apple/swift-nio-zlib-support.git", from: "1.0.0")
+  .package(url: "https://github.com/apple/swift-nio-zlib-support.git", from: "1.0.0"),
+  .package(url: "https://github.com/apple/swift-nio.git", from: "1.9.0"),
+  .package(url: "https://github.com/apple/swift-nio-nghttp2-support.git", from: "1.0.0"),
+  .package(url: "https://github.com/lukasa/swift-nio-http2.git", .branch("cb-issue-24"))
 ]
 
 var cGRPCDependencies: [Target.Dependency] = []
@@ -35,11 +38,14 @@ let package = Package(
   name: "SwiftGRPC",
   products: [
     .library(name: "SwiftGRPC", targets: ["SwiftGRPC"]),
+    .library(name: "SwiftGRPCNIO", targets: ["SwiftGRPCNIO"]),
   ],
   dependencies: packageDependencies,
   targets: [
     .target(name: "SwiftGRPC",
             dependencies: ["CgRPC", "SwiftProtobuf"]),
+    .target(name: "SwiftGRPCNIO",
+            dependencies: ["SwiftProtobuf", "NIOHTTP1", "NIOFoundationCompat", "NIOHTTP2"]),
     .target(name: "CgRPC",
             dependencies: cGRPCDependencies),
     .target(name: "RootsEncoder"),
@@ -58,7 +64,8 @@ let package = Package(
     .target(name: "Simple",
             dependencies: ["SwiftGRPC", "Commander"],
             path: "Sources/Examples/Simple"),
-    .testTarget(name: "SwiftGRPCTests", dependencies: ["SwiftGRPC"])
+    .testTarget(name: "SwiftGRPCTests", dependencies: ["SwiftGRPC"]),
+    .testTarget(name: "SwiftGRPCNIOTests", dependencies: ["SwiftGRPC", "SwiftGRPCNIO"])
   ],
   cLanguageStandard: .gnu11,
   cxxLanguageStandard: .cxx11)

Sources/SwiftGRPCNIO/CallHandlers/BaseCallHandler.swift

@@ -0,0 +1,27 @@
+import Foundation
+import SwiftProtobuf
+import NIO
+import NIOHTTP1
+
+// Provides a means for decoding incoming gRPC messages into protobuf objects.
+// Calls through to `processMessage` for individual messages it receives, which needs to be implemented by subclasses.
+public class BaseCallHandler<RequestMessage: Message, ResponseMessage: Message>: GRPCCallHandler, ChannelInboundHandler {
+  public func makeGRPCServerCodec() -> ChannelHandler { return GRPCServerCodec<RequestMessage, ResponseMessage>() }
+
+  public typealias InboundIn = GRPCServerRequestPart<RequestMessage>
+  public typealias OutboundOut = GRPCServerResponsePart<ResponseMessage>
+
+  public func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
+    switch self.unwrapInboundIn(data) {
+    case .head: preconditionFailure("should not have received headers")
+    case .message(let message): processMessage(message)
+    case .end: endOfStreamReceived()
+    }
+  }
+
+  public func processMessage(_ message: RequestMessage) {
+    fatalError("needs to be overridden")
+  }
+
+  public func endOfStreamReceived() { }
+}

Sources/SwiftGRPCNIO/CallHandlers/BidirectionalStreamingCallHandler.swift

@@ -0,0 +1,42 @@
+import Foundation
+import SwiftProtobuf
+import NIO
+import NIOHTTP1
+
+// Handles bidirectional streaming calls. Forwards incoming messages and end-of-stream events to the observer block.
+// The observer block is implemented by the framework user and calls `context.sendResponse` as needed.
+// To close the call and send the status, fulfill `context.statusPromise`.
+public class BidirectionalStreamingCallHandler<RequestMessage: Message, ResponseMessage: Message>: BaseCallHandler<RequestMessage, ResponseMessage> {
+  public typealias EventObserver = (StreamEvent<RequestMessage>) -> Void
+  fileprivate var eventObserver: EventLoopFuture<EventObserver>?
+
+  //! FIXME: Do we need to keep the context around at all here?
+  public private(set) var context: StreamingResponseCallContext<ResponseMessage>?
+
+  // 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.
+  public init(channel: Channel, request: HTTPRequestHead, eventObserverFactory: (StreamingResponseCallContext<ResponseMessage>) -> EventLoopFuture<EventObserver>) {
+    super.init()
+    self.context = StreamingResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    self.eventObserver = eventObserverFactory(context!)
+    // Terminate the call if no observer is provided.
+    self.eventObserver?.cascadeFailure(promise: context!.statusPromise)
+    context!.statusPromise.futureResult.whenComplete {
+      // When done, reset references to avoid retain cycles.
+      self.eventObserver = nil
+      self.context = nil
+    }
+  }
+
+  public override func processMessage(_ message: RequestMessage) {
+    eventObserver?.whenSuccess { observer in
+      observer(.message(message))
+    }
+  }
+
+  public override func endOfStreamReceived() {
+    eventObserver?.whenSuccess { observer in
+      observer(.end)
+    }
+  }
+}

Sources/SwiftGRPCNIO/CallHandlers/ClientStreamingCallHandler.swift

@@ -0,0 +1,41 @@
+import Foundation
+import SwiftProtobuf
+import NIO
+import NIOHTTP1
+
+// Handles client-streaming calls. Forwards incoming messages and end-of-stream events to the observer block.
+// The observer block is implemented by the framework user and fulfills `context.responsePromise` when done.
+public class ClientStreamingCallHandler<RequestMessage: Message, ResponseMessage: Message>: BaseCallHandler<RequestMessage, ResponseMessage> {
+  public typealias EventObserver = (StreamEvent<RequestMessage>) -> Void
+  fileprivate var eventObserver: EventLoopFuture<EventObserver>?
+
+  //! FIXME: Do we need to keep the context around at all here?
+  public private(set) var context: UnaryResponseCallContext<ResponseMessage>?
+
+  // 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.
+  public init(channel: Channel, request: HTTPRequestHead, eventObserverFactory: (UnaryResponseCallContext<ResponseMessage>) -> EventLoopFuture<EventObserver>) {
+    super.init()
+    self.context = UnaryResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    self.eventObserver = eventObserverFactory(context!)
+    // Terminate the call if no observer is provided.
+    self.eventObserver!.cascadeFailure(promise: context!.responsePromise)
+    context!.responsePromise.futureResult.whenComplete {
+      // When done, reset references to avoid retain cycles.
+      self.eventObserver = nil
+      self.context = nil
+    }
+  }
+
+  public override func processMessage(_ message: RequestMessage) {
+    eventObserver?.whenSuccess { observer in
+      observer(.message(message))
+    }
+  }
+
+  public override func endOfStreamReceived() {
+    eventObserver?.whenSuccess { observer in
+      observer(.end)
+    }
+  }
+}

Sources/SwiftGRPCNIO/CallHandlers/ServerStreamingCallHandler.swift

@@ -0,0 +1,43 @@
+import Foundation
+import SwiftProtobuf
+import NIO
+import NIOHTTP1
+
+// Handles server-streaming calls. Calls the observer block with the request message.
+// 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 class ServerStreamingCallHandler<RequestMessage: Message, ResponseMessage: Message>: BaseCallHandler<RequestMessage, ResponseMessage> {
+  public typealias EventObserver = (RequestMessage) -> EventLoopFuture<GRPCStatus>
+  fileprivate var eventObserver: EventObserver?
+
+  fileprivate var hasReceivedRequest = false
+
+  //! FIXME: Do we need to keep the context around at all here?
+  public private(set) var context: StreamingResponseCallContext<ResponseMessage>?
+
+  public init(channel: Channel, request: HTTPRequestHead, eventObserverFactory: (StreamingResponseCallContext<ResponseMessage>) -> EventObserver) {
+    super.init()
+    self.context = StreamingResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    self.eventObserver = eventObserverFactory(context!)
+    context!.statusPromise.futureResult.whenComplete {
+      // When done, reset references to avoid retain cycles.
+      self.eventObserver = nil
+      self.context = nil
+    }
+  }
+
+
+  public override func processMessage(_ message: RequestMessage) {
+    guard !hasReceivedRequest else {
+      //! FIXME: Better handle this error.
+      print("multiple messages received on server-streaming call")
+      return
+    }
+    hasReceivedRequest = true
+
+    let resultFuture = self.eventObserver!(message)
+    resultFuture
+      // Fulfill the status promise with whatever status the framework user has provided.
+      .cascade(promise: context!.statusPromise)
+  }
+}

Sources/SwiftGRPCNIO/CallHandlers/UnaryCallHandler.swift

@@ -0,0 +1,43 @@
+import Foundation
+import SwiftProtobuf
+import NIO
+import NIOHTTP1
+
+// Handles unary calls. Calls the observer block with the request message.
+// The observer block is implemented by the framework user and returns a future containing the call result.
+// To return a response to the client, the framework user should complete that framework
+// (similar to e.g. serving regular HTTP requests in frameworks such as Vapor).
+public class UnaryCallHandler<RequestMessage: Message, ResponseMessage: Message>: BaseCallHandler<RequestMessage, ResponseMessage> {
+  public typealias EventObserver = (RequestMessage) -> EventLoopFuture<ResponseMessage>
+  fileprivate var eventObserver: EventObserver?
+
+  fileprivate var hasReceivedRequest = false
+
+  //! FIXME: Do we need to keep the context around at all here?
+  public private(set) var context: UnaryResponseCallContext<ResponseMessage>?
+
+  public init(channel: Channel, request: HTTPRequestHead, eventObserverFactory: (UnaryResponseCallContext<ResponseMessage>) -> EventObserver) {
+    super.init()
+    self.context = UnaryResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    self.eventObserver = eventObserverFactory(self.context!)
+    context!.responsePromise.futureResult.whenComplete {
+      // When done, reset references to avoid retain cycles.
+      self.eventObserver = nil
+      self.context = nil
+    }
+  }
+
+  public override func processMessage(_ message: RequestMessage) {
+    guard !hasReceivedRequest else {
+      //! FIXME: Better handle this error.
+      print("multiple messages received on unary call")
+      return
+    }
+    hasReceivedRequest = true
+
+    let resultFuture = self.eventObserver!(message)
+    resultFuture
+      // Fulfill the response promise with whatever response (or error) the framework user has provided.
+      .cascade(promise: context!.responsePromise)
+  }
+}

Sources/SwiftGRPCNIO/GRPCChannelHandler.swift

@@ -0,0 +1,60 @@
+import Foundation
+import SwiftProtobuf
+import NIO
+import NIOHTTP1
+
+// Processes individual gRPC messages and stream-close events on a HTTP2 channel.
+public protocol GRPCCallHandler: ChannelHandler {
+  func makeGRPCServerCodec() -> ChannelHandler
+}
+
+// Provides `GRPCCallHandler` objects for the methods on a particular service name.
+// Implemented by the generated code.
+public protocol CallHandlerProvider {
+  var serviceName: String { get }
+
+  // Looks up and returns the `GRPCCallHandler` for a particular method. Returns nil for unsupported methods.
+  func handleMethod(_ methodName: String, request: HTTPRequestHead, serverHandler: GRPCChannelHandler, channel: Channel) -> GRPCCallHandler?
+}
+
+// Listens on a newly-opened HTTP2 subchannel and waits for the request headers to become available.
+// Once those are available, asks the `CallHandlerProvider` corresponding to the request's service name for an
+// `GRPCCallHandler` object. That object is then forwarded the individual gRPC messages.
+public final class GRPCChannelHandler: ChannelInboundHandler {
+  public typealias InboundIn = RawGRPCServerRequestPart
+
+  public typealias OutboundOut = RawGRPCServerResponsePart
+
+  fileprivate let servicesByName: [String: CallHandlerProvider]
+
+  public init(servicesByName: [String: CallHandlerProvider]) {
+    self.servicesByName = servicesByName
+  }
+
+  public func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
+    let requestPart = self.unwrapInboundIn(data)
+    switch requestPart {
+    case .head(let requestHead):
+      let uriComponents = requestHead.uri.components(separatedBy: "/")
+      guard uriComponents.count >= 3 && uriComponents[0].isEmpty,
+        let providerForServiceName = servicesByName[uriComponents[1]],
+        let callHandler = providerForServiceName.handleMethod(uriComponents[2], request: requestHead, serverHandler: self, channel: ctx.channel) else {
+          ctx.writeAndFlush(self.wrapOutboundOut(.status(.unimplemented(method: requestHead.uri))), promise: nil)
+          return
+      }
+
+      var responseHeaders = HTTPHeaders()
+      responseHeaders.add(name: "content-type", value: "application/grpc")
+      ctx.write(self.wrapOutboundOut(.headers(responseHeaders)), promise: nil)
+
+      let codec = callHandler.makeGRPCServerCodec()
+      ctx.pipeline.add(handler: codec, after: self)
+        .then { ctx.pipeline.add(handler: callHandler, after: codec) }
+        //! FIXME(lukasa): Fix the ordering of this with NIO 1.12 and replace with `remove(, promise:)`.
+        .whenComplete { _ = ctx.pipeline.remove(handler: self) }
+
+    case .message, .end:
+      preconditionFailure("received \(requestPart), should have been removed as a handler at this point")
+    }
+  }
+}

Sources/SwiftGRPCNIO/GRPCServer.swift

@@ -0,0 +1,56 @@
+import Foundation
+import NIO
+import NIOHTTP1
+import NIOHTTP2
+
+// Wrapper object to manage the lifecycle of a gRPC server.
+public final class GRPCServer {
+  public static func start(
+    hostname: String,
+    port: Int,
+    eventLoopGroup: EventLoopGroup,
+    serviceProviders: [CallHandlerProvider]) -> EventLoopFuture<GRPCServer> {
+    let servicesByName = Dictionary(uniqueKeysWithValues: serviceProviders.map { ($0.serviceName, $0) })
+    let bootstrap = ServerBootstrap(group: eventLoopGroup)
+      // Specify backlog and enable SO_REUSEADDR for the server itself
+      .serverChannelOption(ChannelOptions.backlog, value: 256)
+      .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
+
+      // Set the handlers that are applied to the accepted Channels
+      .childChannelInitializer { channel in
+        //! FIXME: Add an option for gRPC-via-HTTP1 (pPRC).
+        return channel.pipeline.add(handler: HTTP2Parser(mode: .server)).then {
+          let multiplexer = HTTP2StreamMultiplexer { (channel, streamID) -> EventLoopFuture<Void> in
+            return channel.pipeline.add(handler: HTTP2ToHTTP1ServerCodec(streamID: streamID))
+              .then { channel.pipeline.add(handler: HTTP1ToRawGRPCServerCodec()) }
+              .then { channel.pipeline.add(handler: GRPCChannelHandler(servicesByName: servicesByName)) }
+          }
+
+          return channel.pipeline.add(handler: multiplexer)
+        }
+      }
+
+      // Enable TCP_NODELAY and SO_REUSEADDR for the accepted Channels
+      .childChannelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
+      .childChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
+      .childChannelOption(ChannelOptions.maxMessagesPerRead, value: 1)
+      .childChannelOption(ChannelOptions.allowRemoteHalfClosure, value: true)
+
+    return bootstrap.bind(host: hostname, port: port)
+      .map { GRPCServer(channel: $0) }
+  }
+
+  fileprivate let channel: Channel
+
+  fileprivate init(channel: Channel) {
+    self.channel = channel
+  }
+
+  public var onClose: EventLoopFuture<Void> {
+    return channel.closeFuture
+  }
+
+  public func close() -> EventLoopFuture<Void> {
+    return channel.close(mode: .all)
+  }
+}