Merge pull request #1366 from glbrntt/gb-update-async-await

Merge tag '1.7.1' into 1.7.1-async-await

.github/workflows/ci.yaml

@@ -3,7 +3,7 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main, 1.6.0-async-await]
+    branches: [main, 1.7.1-async-await]
 jobs:
   preflight:
     name: License Header and Formatting Checks

CGRPCZlib.podspec

@@ -2,7 +2,7 @@ Pod::Spec.new do |s|
 
     s.name = 'CGRPCZlib'
     s.module_name = 'CGRPCZlib'
-    s.version = '1.5.0'
+    s.version = '1.6.0'
     s.license = { :type => 'Apache 2.0', :file => 'LICENSE' }
     s.summary = 'Compression library that provides in-memory compression and decompression functions'
     s.homepage = 'https://www.grpc.io'

Package.swift

@@ -36,7 +36,7 @@ let packageDependencies: [Package.Dependency] = [
   ),
   .package(
     url: "https://github.com/apple/swift-nio-http2.git",
-    from: "1.18.2"
+    from: "1.19.2"
   ),
   .package(
     url: "https://github.com/apple/swift-nio-transport-services.git",
@@ -232,6 +232,7 @@ extension Target {
     name: "GRPCPerformanceTests",
     dependencies: [
       .grpc,
+      .grpcSampleData,
       .nioCore,
       .nioEmbedded,
       .nioPosix,

README.md

@@ -130,9 +130,15 @@ the following line to your `Podfile`:
 ```ruby
     pod 'gRPC-Swift-Plugins'
 ```
-
 The plugins are available in the `Pods/gRPC-Swift-Plugins/` folder afterwards.
 
+#### Homebrew
+
+The plugins are available from [homebrew](https://brew.sh) and can be installed with:
+```bash
+    $ brew install swift-protobuf grpc-swift
+```
+
 ## Examples
 
 gRPC Swift has a number of tutorials and examples available. They are split

Sources/GRPC/CallOptions.swift

@@ -133,6 +133,7 @@ extension CallOptions {
       self.source = source
     }
 
+    @usableFromInline
     internal func requestID() -> String? {
       switch self.source {
       case .none:

Sources/GRPC/ConnectionManager.swift

@@ -591,7 +591,7 @@ internal final class ConnectionManager {
       ])
 
     case .connecting:
-      self.invalidState()
+      self.connectionFailed(withError: error)
 
     case var .active(state):
       state.error = error

Sources/GRPC/ConnectionPool/PooledChannel.swift

@@ -132,6 +132,11 @@ internal final class PooledChannel: GRPCChannel {
     callOptions: CallOptions,
     interceptors: [ClientInterceptor<Request, Response>]
   ) -> Call<Request, Response> where Request: Message, Response: Message {
+    var callOptions = callOptions
+    if let requestID = callOptions.requestIDProvider.requestID() {
+      callOptions.applyRequestID(requestID)
+    }
+
     let (stream, eventLoop) = self._makeStreamChannel(callOptions: callOptions)
 
     return Call(
@@ -157,6 +162,11 @@ internal final class PooledChannel: GRPCChannel {
     callOptions: CallOptions,
     interceptors: [ClientInterceptor<Request, Response>]
   ) -> Call<Request, Response> where Request: GRPCPayload, Response: GRPCPayload {
+    var callOptions = callOptions
+    if let requestID = callOptions.requestIDProvider.requestID() {
+      callOptions.applyRequestID(requestID)
+    }
+
     let (stream, eventLoop) = self._makeStreamChannel(callOptions: callOptions)
 
     return Call(
@@ -192,3 +202,14 @@ internal final class PooledChannel: GRPCChannel {
     self._pool.shutdown(mode: .graceful(deadline), promise: promise)
   }
 }
+
+extension CallOptions {
+  @usableFromInline
+  mutating func applyRequestID(_ requestID: String) {
+    self.logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
+    // Add the request ID header too.
+    if let requestIDHeader = self.requestIDHeader {
+      self.customMetadata.add(name: requestIDHeader, value: requestID)
+    }
+  }
+}

Sources/GRPC/GRPCStatus.swift

@@ -105,10 +105,15 @@ public struct GRPCStatus: Error {
   ///   status code. Use `GRPCStatus.isOk` or check the code directly.
   public static let ok = GRPCStatus(code: .ok, message: nil)
   /// "Internal server error" status.
-  public static let processingError = GRPCStatus(
-    code: .internalError,
-    message: "unknown error processing request"
-  )
+  public static let processingError = Self.processingError(cause: nil)
+
+  public static func processingError(cause: Error?) -> GRPCStatus {
+    return GRPCStatus(
+      code: .internalError,
+      message: "unknown error processing request",
+      cause: cause
+    )
+  }
 }
 
 extension GRPCStatus: Equatable {
@@ -119,9 +124,14 @@ extension GRPCStatus: Equatable {
 
 extension GRPCStatus: CustomStringConvertible {
   public var description: String {
-    if let message = message {
+    switch (self.message, self.cause) {
+    case let (.some(message), .some(cause)):
+      return "\(self.code): \(message), cause: \(cause)"
+    case let (.some(message), .none):
       return "\(self.code): \(message)"
-    } else {
+    case let (.none, .some(cause)):
+      return "\(self.code), cause: \(cause)"
+    case (.none, .none):
       return "\(self.code)"
     }
   }

Sources/GRPC/Interceptor/ClientTransport.swift

@@ -850,7 +850,7 @@ extension ClientTransport {
     promise: EventLoopPromise<Void>?
   ) {
     self.callEventLoop.assertInEventLoop()
-    self.logger.debug("buffering request part", metadata: [
+    self.logger.trace("buffering request part", metadata: [
       "request_part": "\(part.name)",
       "call_state": self.stateForLogging,
     ])
@@ -868,7 +868,7 @@ extension ClientTransport {
     // Save any flushing until we're done writing.
     var shouldFlush = false
 
-    self.logger.debug("unbuffering request parts", metadata: [
+    self.logger.trace("unbuffering request parts", metadata: [
       "request_parts": "\(self.writeBuffer.count)",
     ])
 
@@ -878,7 +878,7 @@ extension ClientTransport {
     while self.state.isUnbuffering, !self.writeBuffer.isEmpty {
       // Pull out as many writes as possible.
       while let write = self.writeBuffer.popFirst() {
-        self.logger.debug("unbuffering request part", metadata: [
+        self.logger.trace("unbuffering request part", metadata: [
           "request_part": "\(write.request.name)",
         ])
 
@@ -897,7 +897,7 @@ extension ClientTransport {
     }
 
     if self.writeBuffer.isEmpty {
-      self.logger.debug("request buffer drained")
+      self.logger.trace("request buffer drained")
     } else {
       self.logger.notice("unbuffering aborted", metadata: ["call_state": self.stateForLogging])
     }
@@ -914,7 +914,7 @@ extension ClientTransport {
   /// Fails any promises that come with buffered writes with `error`.
   /// - Parameter error: The `Error` to fail promises with.
   private func failBufferedWrites(with error: Error) {
-    self.logger.debug("failing buffered writes", metadata: ["call_state": self.stateForLogging])
+    self.logger.trace("failing buffered writes", metadata: ["call_state": self.stateForLogging])
 
     while let write = self.writeBuffer.popFirst() {
       write.promise?.fail(error)

Sources/GRPC/ServerErrorProcessor.swift

@@ -44,7 +44,7 @@ internal enum ServerErrorProcessor {
       trailers = [:]
     } else {
       // Eh... well, we don't what status to use. Use a generic one.
-      status = .processingError
+      status = .processingError(cause: error)
       trailers = [:]
     }
 
@@ -84,7 +84,7 @@ internal enum ServerErrorProcessor {
       mergedTrailers = trailers
     } else {
       // Eh... well, we don't what status to use. Use a generic one.
-      status = .processingError
+      status = .processingError(cause: error)
       mergedTrailers = trailers
     }
 

Sources/GRPC/Version.swift

@@ -19,10 +19,10 @@ internal enum Version {
   internal static let major = 1
 
   /// The minor version.
-  internal static let minor = 6
+  internal static let minor = 7
 
   /// The patch version.
-  internal static let patch = 0
+  internal static let patch = 1
 
   /// The version string.
   internal static let versionString = "\(major).\(minor).\(patch)-async-await.1"

Sources/GRPCPerformanceTests/Benchmarks/ServerProvidingBenchmark.swift

@@ -14,26 +14,56 @@
  * limitations under the License.
  */
 import GRPC
+import GRPCSampleData
 import NIOCore
 import NIOPosix
 
 class ServerProvidingBenchmark: Benchmark {
   private let providers: [CallHandlerProvider]
   private let threadCount: Int
+  private let useNIOTSIfAvailable: Bool
+  private let useTLS: Bool
   private var group: EventLoopGroup!
   private(set) var server: Server!
 
-  init(providers: [CallHandlerProvider], threadCount: Int = 1) {
+  init(
+    providers: [CallHandlerProvider],
+    useNIOTSIfAvailable: Bool,
+    useTLS: Bool,
+    threadCount: Int = 1
+  ) {
     self.providers = providers
+    self.useNIOTSIfAvailable = useNIOTSIfAvailable
+    self.useTLS = useTLS
     self.threadCount = threadCount
   }
 
   func setUp() throws {
-    self.group = MultiThreadedEventLoopGroup(numberOfThreads: self.threadCount)
-    self.server = try Server.insecure(group: self.group)
-      .withServiceProviders(self.providers)
-      .bind(host: "127.0.0.1", port: 0)
-      .wait()
+    if self.useNIOTSIfAvailable {
+      self.group = PlatformSupport.makeEventLoopGroup(loopCount: self.threadCount)
+    } else {
+      self.group = MultiThreadedEventLoopGroup(numberOfThreads: self.threadCount)
+    }
+
+    if self.useTLS {
+      #if canImport(NIOSSL)
+      self.server = try Server.usingTLSBackedByNIOSSL(
+        on: self.group,
+        certificateChain: [SampleCertificate.server.certificate],
+        privateKey: SamplePrivateKey.server
+      ).withTLS(trustRoots: .certificates([SampleCertificate.ca.certificate]))
+        .withServiceProviders(self.providers)
+        .bind(host: "127.0.0.1", port: 0)
+        .wait()
+      #else
+      fatalError("NIOSSL must be imported to use TLS")
+      #endif
+    } else {
+      self.server = try Server.insecure(group: self.group)
+        .withServiceProviders(self.providers)
+        .bind(host: "127.0.0.1", port: 0)
+        .wait()
+    }
   }
 
   func tearDown() throws {

Sources/GRPCPerformanceTests/Benchmarks/UnaryThroughput.swift

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 import GRPC
+import GRPCSampleData
 import NIOCore
 import NIOPosix
 
@@ -22,23 +23,50 @@ import NIOPosix
 /// Requests are sent in batches of (up-to) 100 requests. This is due to
 /// https://github.com/apple/swift-nio-http2/issues/87#issuecomment-483542401.
 class Unary: ServerProvidingBenchmark {
+  private let useNIOTSIfAvailable: Bool
+  private let useTLS: Bool
   private var group: EventLoopGroup!
   private(set) var client: Echo_EchoClient!
 
   let requestCount: Int
   let requestText: String
 
-  init(requests: Int, text: String) {
+  init(requests: Int, text: String, useNIOTSIfAvailable: Bool, useTLS: Bool) {
+    self.useNIOTSIfAvailable = useNIOTSIfAvailable
+    self.useTLS = useTLS
     self.requestCount = requests
     self.requestText = text
-    super.init(providers: [MinimalEchoProvider()])
+    super.init(
+      providers: [MinimalEchoProvider()],
+      useNIOTSIfAvailable: useNIOTSIfAvailable,
+      useTLS: useTLS
+    )
   }
 
   override func setUp() throws {
     try super.setUp()
-    self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
-    let channel = ClientConnection.insecure(group: self.group)
-      .connect(host: "127.0.0.1", port: self.server.channel.localAddress!.port!)
+
+    if self.useNIOTSIfAvailable {
+      self.group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+    } else {
+      self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    }
+
+    let channel: ClientConnection
+
+    if self.useTLS {
+      #if canImport(NIOSSL)
+      channel = ClientConnection.usingTLSBackedByNIOSSL(on: self.group)
+        .withTLS(trustRoots: .certificates([SampleCertificate.ca.certificate]))
+        .withTLS(serverHostnameOverride: "localhost")
+        .connect(host: "127.0.0.1", port: self.server.channel.localAddress!.port!)
+      #else
+      fatalError("NIOSSL must be imported to use TLS")
+      #endif
+    } else {
+      channel = ClientConnection.insecure(group: self.group)
+        .connect(host: "127.0.0.1", port: self.server.channel.localAddress!.port!)
+    }
 
     self.client = .init(channel: channel)
   }
@@ -72,9 +100,14 @@ class Unary: ServerProvidingBenchmark {
 class Bidi: Unary {
   let batchSize: Int
 
-  init(requests: Int, text: String, batchSize: Int) {
+  init(requests: Int, text: String, batchSize: Int, useNIOTSIfAvailable: Bool, useTLS: Bool) {
     self.batchSize = batchSize
-    super.init(requests: requests, text: text)
+    super.init(
+      requests: requests,
+      text: text,
+      useNIOTSIfAvailable: useNIOTSIfAvailable,
+      useTLS: useTLS
+    )
   }
 
   override func run() throws -> Int {