ClientConnection.swift 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725
  1. /*
  2. * Copyright 2019, gRPC Authors All rights reserved.
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. import Logging
  17. import NIOCore
  18. import NIOHPACK
  19. import NIOHTTP2
  20. import NIOPosix
  21. import NIOTLS
  22. import NIOTransportServices
  23. import SwiftProtobuf
  24. #if os(Linux)
  25. @preconcurrency import Foundation
  26. #else
  27. import Foundation
  28. #endif
  29. #if canImport(NIOSSL)
  30. import NIOSSL
  31. #endif
  32. #if canImport(Network)
  33. import Network
  34. #endif
  35. /// Provides a single, managed connection to a server which is guaranteed to always use the same
  36. /// `EventLoop`.
  37. ///
  38. /// The connection to the server is provided by a single channel which will attempt to reconnect to
  39. /// the server if the connection is dropped. When either the client or server detects that the
  40. /// connection has become idle -- that is, there are no outstanding RPCs and the idle timeout has
  41. /// passed (5 minutes, by default) -- the underlying channel will be closed. The client will not
  42. /// idle the connection if any RPC exists, even if there has been no activity on the RPC for the
  43. /// idle timeout. Long-lived, low activity RPCs may benefit from configuring keepalive (see
  44. /// ``ClientConnectionKeepalive``) which periodically pings the server to ensure that the connection
  45. /// is not dropped. If the connection is idle a new channel will be created on-demand when the next
  46. /// RPC is made.
  47. ///
  48. /// The state of the connection can be observed using a ``ConnectivityStateDelegate``.
  49. ///
  50. /// Since the connection is managed, and may potentially spend long periods of time waiting for a
  51. /// connection to come up (cellular connections, for example), different behaviors may be used when
  52. /// starting a call. The different behaviors are detailed in the ``CallStartBehavior`` documentation.
  53. ///
  54. /// ### Channel Pipeline
  55. ///
  56. /// The `NIO.ChannelPipeline` for the connection is configured as such:
  57. ///
  58. /// ┌──────────────────────────┐
  59. /// │ DelegatingErrorHandler │
  60. /// └──────────▲───────────────┘
  61. /// HTTP2Frame│
  62. /// │ ⠇ ⠇ ⠇ ⠇
  63. /// │ ┌┴─▼┐ ┌┴─▼┐
  64. /// │ │ | │ | HTTP/2 streams
  65. /// │ └▲─┬┘ └▲─┬┘
  66. /// │ │ │ │ │ HTTP2Frame
  67. /// ┌─┴────────────────┴─▼───┴─▼┐
  68. /// │ HTTP2StreamMultiplexer |
  69. /// └─▲───────────────────────┬─┘
  70. /// HTTP2Frame│ │HTTP2Frame
  71. /// ┌─┴───────────────────────▼─┐
  72. /// │ GRPCIdleHandler │
  73. /// └─▲───────────────────────┬─┘
  74. /// HTTP2Frame│ │HTTP2Frame
  75. /// ┌─┴───────────────────────▼─┐
  76. /// │ NIOHTTP2Handler │
  77. /// └─▲───────────────────────┬─┘
  78. /// ByteBuffer│ │ByteBuffer
  79. /// ┌─┴───────────────────────▼─┐
  80. /// │ NIOSSLHandler │
  81. /// └─▲───────────────────────┬─┘
  82. /// ByteBuffer│ │ByteBuffer
  83. /// │ ▼
  84. ///
  85. /// The 'GRPCIdleHandler' intercepts HTTP/2 frames and various events and is responsible for
  86. /// informing and controlling the state of the connection (idling and keepalive). The HTTP/2 streams
  87. /// are used to handle individual RPCs.
  88. public final class ClientConnection: Sendable {
  89. private let connectionManager: ConnectionManager
  90. /// HTTP multiplexer from the underlying channel handling gRPC calls.
  91. internal func getMultiplexer() -> EventLoopFuture<HTTP2StreamMultiplexer> {
  92. return self.connectionManager.getHTTP2Multiplexer()
  93. }
  94. /// The configuration for this client.
  95. internal let configuration: Configuration
  96. /// The scheme of the URI for each RPC, i.e. 'http' or 'https'.
  97. internal let scheme: String
  98. /// The authority of the URI for each RPC.
  99. internal let authority: String
  100. /// A monitor for the connectivity state.
  101. public let connectivity: ConnectivityStateMonitor
  102. /// The `EventLoop` this connection is using.
  103. public var eventLoop: EventLoop {
  104. return self.connectionManager.eventLoop
  105. }
  106. /// Creates a new connection from the given configuration. Prefer using
  107. /// ``ClientConnection/secure(group:)`` to build a connection secured with TLS or
  108. /// ``ClientConnection/insecure(group:)`` to build a plaintext connection.
  109. ///
  110. /// - Important: Users should prefer using ``ClientConnection/secure(group:)`` to build a connection
  111. /// with TLS, or ``ClientConnection/insecure(group:)`` to build a connection without TLS.
  112. public init(configuration: Configuration) {
  113. self.configuration = configuration
  114. self.scheme = configuration.tlsConfiguration == nil ? "http" : "https"
  115. self.authority = configuration.tlsConfiguration?.hostnameOverride ?? configuration.target.host
  116. let monitor = ConnectivityStateMonitor(
  117. delegate: configuration.connectivityStateDelegate,
  118. queue: configuration.connectivityStateDelegateQueue
  119. )
  120. self.connectivity = monitor
  121. self.connectionManager = ConnectionManager(
  122. configuration: configuration,
  123. connectivityDelegate: monitor,
  124. idleBehavior: .closeWhenIdleTimeout,
  125. logger: configuration.backgroundActivityLogger
  126. )
  127. }
  128. /// Close the channel, and any connections associated with it. Any ongoing RPCs may fail.
  129. ///
  130. /// - Returns: Returns a future which will be resolved when shutdown has completed.
  131. public func close() -> EventLoopFuture<Void> {
  132. let promise = self.eventLoop.makePromise(of: Void.self)
  133. self.close(promise: promise)
  134. return promise.futureResult
  135. }
  136. /// Close the channel, and any connections associated with it. Any ongoing RPCs may fail.
  137. ///
  138. /// - Parameter promise: A promise which will be completed when shutdown has completed.
  139. public func close(promise: EventLoopPromise<Void>) {
  140. self.connectionManager.shutdown(mode: .forceful, promise: promise)
  141. }
  142. /// Attempt to gracefully shutdown the channel. New RPCs will be failed immediately and existing
  143. /// RPCs may continue to run until they complete.
  144. ///
  145. /// - Parameters:
  146. /// - deadline: A point in time by which the graceful shutdown must have completed. If the
  147. /// deadline passes and RPCs are still active then the connection will be closed forcefully
  148. /// and any remaining in-flight RPCs may be failed.
  149. /// - promise: A promise which will be completed when shutdown has completed.
  150. public func closeGracefully(deadline: NIODeadline, promise: EventLoopPromise<Void>) {
  151. return self.connectionManager.shutdown(mode: .graceful(deadline), promise: promise)
  152. }
  153. /// Populates the logger in `options` and appends a request ID header to the metadata, if
  154. /// configured.
  155. /// - Parameter options: The options containing the logger to populate.
  156. private func populateLogger(in options: inout CallOptions) {
  157. // Get connection metadata.
  158. self.connectionManager.appendMetadata(to: &options.logger)
  159. // Attach a request ID.
  160. let requestID = options.requestIDProvider.requestID()
  161. if let requestID = requestID {
  162. options.logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
  163. // Add the request ID header too.
  164. if let requestIDHeader = options.requestIDHeader {
  165. options.customMetadata.add(name: requestIDHeader, value: requestID)
  166. }
  167. }
  168. }
  169. }
  170. extension ClientConnection: GRPCChannel {
  171. public func makeCall<Request: Message, Response: Message>(
  172. path: String,
  173. type: GRPCCallType,
  174. callOptions: CallOptions,
  175. interceptors: [ClientInterceptor<Request, Response>]
  176. ) -> Call<Request, Response> {
  177. var options = callOptions
  178. self.populateLogger(in: &options)
  179. let multiplexer = self.getMultiplexer()
  180. let eventLoop = callOptions.eventLoopPreference.exact ?? multiplexer.eventLoop
  181. // This should be on the same event loop as the multiplexer (i.e. the event loop of the
  182. // underlying `Channel`.
  183. let channel = multiplexer.eventLoop.makePromise(of: Channel.self)
  184. multiplexer.whenComplete {
  185. ClientConnection.makeStreamChannel(using: $0, promise: channel)
  186. }
  187. return Call(
  188. path: path,
  189. type: type,
  190. eventLoop: eventLoop,
  191. options: options,
  192. interceptors: interceptors,
  193. transportFactory: .http2(
  194. channel: channel.futureResult,
  195. authority: self.authority,
  196. scheme: self.scheme,
  197. maximumReceiveMessageLength: self.configuration.maximumReceiveMessageLength,
  198. errorDelegate: self.configuration.errorDelegate
  199. )
  200. )
  201. }
  202. public func makeCall<Request: GRPCPayload, Response: GRPCPayload>(
  203. path: String,
  204. type: GRPCCallType,
  205. callOptions: CallOptions,
  206. interceptors: [ClientInterceptor<Request, Response>]
  207. ) -> Call<Request, Response> {
  208. var options = callOptions
  209. self.populateLogger(in: &options)
  210. let multiplexer = self.getMultiplexer()
  211. let eventLoop = callOptions.eventLoopPreference.exact ?? multiplexer.eventLoop
  212. // This should be on the same event loop as the multiplexer (i.e. the event loop of the
  213. // underlying `Channel`.
  214. let channel = multiplexer.eventLoop.makePromise(of: Channel.self)
  215. multiplexer.whenComplete {
  216. ClientConnection.makeStreamChannel(using: $0, promise: channel)
  217. }
  218. return Call(
  219. path: path,
  220. type: type,
  221. eventLoop: eventLoop,
  222. options: options,
  223. interceptors: interceptors,
  224. transportFactory: .http2(
  225. channel: channel.futureResult,
  226. authority: self.authority,
  227. scheme: self.scheme,
  228. maximumReceiveMessageLength: self.configuration.maximumReceiveMessageLength,
  229. errorDelegate: self.configuration.errorDelegate
  230. )
  231. )
  232. }
  233. private static func makeStreamChannel(
  234. using result: Result<HTTP2StreamMultiplexer, Error>,
  235. promise: EventLoopPromise<Channel>
  236. ) {
  237. switch result {
  238. case let .success(multiplexer):
  239. multiplexer.createStreamChannel(promise: promise) {
  240. $0.eventLoop.makeSucceededVoidFuture()
  241. }
  242. case let .failure(error):
  243. promise.fail(error)
  244. }
  245. }
  246. }
  247. // MARK: - Configuration structures
  248. /// A target to connect to.
  249. public struct ConnectionTarget: Sendable, Hashable {
  250. internal enum Wrapped: Hashable {
  251. case hostAndPort(String, Int)
  252. case unixDomainSocket(String)
  253. case socketAddress(SocketAddress)
  254. case connectedSocket(NIOBSDSocket.Handle)
  255. case vsockAddress(VsockAddress)
  256. }
  257. internal var wrapped: Wrapped
  258. private init(_ wrapped: Wrapped) {
  259. self.wrapped = wrapped
  260. }
  261. /// The host and port. The port is 443 by default.
  262. public static func host(_ host: String, port: Int = 443) -> ConnectionTarget {
  263. return ConnectionTarget(.hostAndPort(host, port))
  264. }
  265. /// The host and port.
  266. public static func hostAndPort(_ host: String, _ port: Int) -> ConnectionTarget {
  267. return ConnectionTarget(.hostAndPort(host, port))
  268. }
  269. /// The path of a Unix domain socket.
  270. public static func unixDomainSocket(_ path: String) -> ConnectionTarget {
  271. return ConnectionTarget(.unixDomainSocket(path))
  272. }
  273. /// A NIO socket address.
  274. public static func socketAddress(_ address: SocketAddress) -> ConnectionTarget {
  275. return ConnectionTarget(.socketAddress(address))
  276. }
  277. /// A connected NIO socket.
  278. public static func connectedSocket(_ socket: NIOBSDSocket.Handle) -> ConnectionTarget {
  279. return ConnectionTarget(.connectedSocket(socket))
  280. }
  281. /// A vsock socket.
  282. public static func vsockAddress(_ vsockAddress: VsockAddress) -> ConnectionTarget {
  283. return ConnectionTarget(.vsockAddress(vsockAddress))
  284. }
  285. @usableFromInline
  286. var host: String {
  287. switch self.wrapped {
  288. case let .hostAndPort(host, _):
  289. return host
  290. case let .socketAddress(.v4(address)):
  291. return address.host
  292. case let .socketAddress(.v6(address)):
  293. return address.host
  294. case .unixDomainSocket, .socketAddress(.unixDomainSocket), .connectedSocket:
  295. return "localhost"
  296. case let .vsockAddress(address):
  297. return "vsock://\(address.cid)"
  298. }
  299. }
  300. }
  301. /// The connectivity behavior to use when starting an RPC.
  302. public struct CallStartBehavior: Hashable, Sendable {
  303. internal enum Behavior: Hashable, Sendable {
  304. case waitsForConnectivity
  305. case fastFailure
  306. }
  307. internal var wrapped: Behavior
  308. private init(_ wrapped: Behavior) {
  309. self.wrapped = wrapped
  310. }
  311. /// Waits for connectivity (that is, the 'ready' connectivity state) before attempting to start
  312. /// an RPC. Doing so may involve multiple connection attempts.
  313. ///
  314. /// This is the preferred, and default, behaviour.
  315. public static let waitsForConnectivity = CallStartBehavior(.waitsForConnectivity)
  316. /// The 'fast failure' behaviour is intended for cases where users would rather their RPC failed
  317. /// quickly rather than waiting for an active connection. The behaviour depends on the current
  318. /// connectivity state:
  319. ///
  320. /// - Idle: a connection attempt will be started and the RPC will fail if that attempt fails.
  321. /// - Connecting: a connection attempt is already in progress, the RPC will fail if that attempt
  322. /// fails.
  323. /// - Ready: a connection is already active: the RPC will be started using that connection.
  324. /// - Transient failure: the last connection or connection attempt failed and gRPC is waiting to
  325. /// connect again. The RPC will fail immediately.
  326. /// - Shutdown: the connection is shutdown, the RPC will fail immediately.
  327. public static let fastFailure = CallStartBehavior(.fastFailure)
  328. }
  329. extension ClientConnection {
  330. /// Configuration for a ``ClientConnection``. Users should prefer using one of the
  331. /// ``ClientConnection`` builders: ``ClientConnection/secure(group:)`` or ``ClientConnection/insecure(group:)``.
  332. public struct Configuration: Sendable {
  333. /// The target to connect to.
  334. public var target: ConnectionTarget
  335. /// The event loop group to run the connection on.
  336. public var eventLoopGroup: EventLoopGroup
  337. /// An error delegate which is called when errors are caught. Provided delegates **must not
  338. /// maintain a strong reference to this `ClientConnection`**. Doing so will cause a retain
  339. /// cycle. Defaults to ``LoggingClientErrorDelegate``.
  340. public var errorDelegate: ClientErrorDelegate? = LoggingClientErrorDelegate.shared
  341. /// A delegate which is called when the connectivity state is changed. Defaults to `nil`.
  342. public var connectivityStateDelegate: ConnectivityStateDelegate?
  343. /// The `DispatchQueue` on which to call the connectivity state delegate. If a delegate is
  344. /// provided but the queue is `nil` then one will be created by gRPC. Defaults to `nil`.
  345. public var connectivityStateDelegateQueue: DispatchQueue?
  346. #if canImport(NIOSSL)
  347. /// TLS configuration for this connection. `nil` if TLS is not desired.
  348. ///
  349. /// - Important: `tls` is deprecated; use ``tlsConfiguration`` or one of
  350. /// the ``ClientConnection/usingTLS(with:on:)`` builder functions.
  351. @available(*, deprecated, renamed: "tlsConfiguration")
  352. public var tls: TLS? {
  353. get {
  354. return self.tlsConfiguration?.asDeprecatedClientConfiguration
  355. }
  356. set {
  357. self.tlsConfiguration = newValue.map { .init(transforming: $0) }
  358. }
  359. }
  360. #endif // canImport(NIOSSL)
  361. /// TLS configuration for this connection. `nil` if TLS is not desired.
  362. public var tlsConfiguration: GRPCTLSConfiguration?
  363. /// The connection backoff configuration. If no connection retrying is required then this should
  364. /// be `nil`.
  365. public var connectionBackoff: ConnectionBackoff? = ConnectionBackoff()
  366. /// The connection keepalive configuration.
  367. public var connectionKeepalive = ClientConnectionKeepalive()
  368. /// The amount of time to wait before closing the connection. The idle timeout will start only
  369. /// if there are no RPCs in progress and will be cancelled as soon as any RPCs start.
  370. ///
  371. /// If a connection becomes idle, starting a new RPC will automatically create a new connection.
  372. ///
  373. /// Defaults to 30 minutes.
  374. public var connectionIdleTimeout: TimeAmount = .minutes(30)
  375. /// The maximum allowed age of a connection.
  376. ///
  377. /// If set, no new RPCs will be started on the connection after the connection has been opened
  378. /// for this period of time. Existing RPCs will be allowed to continue and the connection will
  379. /// close once all RPCs on the connection have finished. If this isn't set then connections have
  380. /// no limit on their lifetime.
  381. public var connectionMaxAge: TimeAmount? = nil
  382. /// The behavior used to determine when an RPC should start. That is, whether it should wait for
  383. /// an active connection or fail quickly if no connection is currently available.
  384. ///
  385. /// Defaults to ``CallStartBehavior/waitsForConnectivity``.
  386. public var callStartBehavior: CallStartBehavior = .waitsForConnectivity
  387. /// The HTTP/2 flow control target window size. Defaults to 8MB. Values are clamped between
  388. /// 1 and 2^31-1 inclusive.
  389. public var httpTargetWindowSize = 8 * 1024 * 1024 {
  390. didSet {
  391. self.httpTargetWindowSize = self.httpTargetWindowSize.clamped(to: 1 ... Int(Int32.max))
  392. }
  393. }
  394. /// The HTTP/2 max frame size. Defaults to 16384. Value is clamped between 2^14 and 2^24-1
  395. /// octets inclusive (the minimum and maximum allowable values - HTTP/2 RFC 7540 4.2).
  396. public var httpMaxFrameSize: Int = 16384 {
  397. didSet {
  398. self.httpMaxFrameSize = self.httpMaxFrameSize.clamped(to: 16384 ... 16_777_215)
  399. }
  400. }
  401. /// The HTTP protocol used for this connection.
  402. public var httpProtocol: HTTP2FramePayloadToHTTP1ClientCodec.HTTPProtocol {
  403. return self.tlsConfiguration == nil ? .http : .https
  404. }
  405. /// The maximum size in bytes of a message which may be received from a server. Defaults to 4MB.
  406. public var maximumReceiveMessageLength: Int = 4 * 1024 * 1024 {
  407. willSet {
  408. precondition(newValue >= 0, "maximumReceiveMessageLength must be positive")
  409. }
  410. }
  411. /// The HTTP/2 max number of reset streams. Defaults to 32. Must be non-negative.
  412. public var httpMaxResetStreams: Int = 32 {
  413. willSet {
  414. precondition(newValue >= 0, "httpMaxResetStreams must be non-negative")
  415. }
  416. }
  417. /// A logger for background information (such as connectivity state). A separate logger for
  418. /// requests may be provided in the `CallOptions`.
  419. ///
  420. /// Defaults to a no-op logger.
  421. public var backgroundActivityLogger = Logger(
  422. label: "io.grpc",
  423. factory: { _ in SwiftLogNoOpLogHandler() }
  424. )
  425. /// A channel initializer which will be run after gRPC has initialized each channel. This may be
  426. /// used to add additional handlers to the pipeline and is intended for debugging.
  427. ///
  428. /// - Warning: The initializer closure may be invoked *multiple times*.
  429. @preconcurrency
  430. public var debugChannelInitializer: (@Sendable (Channel) -> EventLoopFuture<Void>)?
  431. #if canImport(Network)
  432. /// A closure allowing to customise the `NWParameters` used when establishing a connection using `NIOTransportServices`.
  433. @available(macOS 10.14, iOS 12.0, watchOS 6.0, tvOS 12.0, *)
  434. public var nwParametersConfigurator: (@Sendable (NWParameters) -> Void)? {
  435. get {
  436. self._nwParametersConfigurator as! (@Sendable (NWParameters) -> Void)?
  437. }
  438. set {
  439. self._nwParametersConfigurator = newValue
  440. }
  441. }
  442. private var _nwParametersConfigurator: (any Sendable)?
  443. #endif
  444. #if canImport(NIOSSL)
  445. /// Create a `Configuration` with some pre-defined defaults. Prefer using
  446. /// `ClientConnection.secure(group:)` to build a connection secured with TLS or
  447. /// `ClientConnection.insecure(group:)` to build a plaintext connection.
  448. ///
  449. /// - Parameter target: The target to connect to.
  450. /// - Parameter eventLoopGroup: The event loop group to run the connection on.
  451. /// - Parameter errorDelegate: The error delegate, defaulting to a delegate which will log only
  452. /// on debug builds.
  453. /// - Parameter connectivityStateDelegate: A connectivity state delegate, defaulting to `nil`.
  454. /// - Parameter connectivityStateDelegateQueue: A `DispatchQueue` on which to call the
  455. /// `connectivityStateDelegate`.
  456. /// - Parameter tls: TLS configuration, defaulting to `nil`.
  457. /// - Parameter connectionBackoff: The connection backoff configuration to use.
  458. /// - Parameter connectionKeepalive: The keepalive configuration to use.
  459. /// - Parameter connectionIdleTimeout: The amount of time to wait before closing the connection, defaulting to 30 minutes.
  460. /// - Parameter callStartBehavior: The behavior used to determine when a call should start in
  461. /// relation to its underlying connection. Defaults to `waitsForConnectivity`.
  462. /// - Parameter httpTargetWindowSize: The HTTP/2 flow control target window size.
  463. /// - Parameter backgroundActivityLogger: A logger for background information (such as
  464. /// connectivity state). Defaults to a no-op logger.
  465. /// - Parameter debugChannelInitializer: A channel initializer will be called after gRPC has
  466. /// initialized the channel. Defaults to `nil`.
  467. @available(*, deprecated, renamed: "default(target:eventLoopGroup:)")
  468. @preconcurrency
  469. public init(
  470. target: ConnectionTarget,
  471. eventLoopGroup: EventLoopGroup,
  472. errorDelegate: ClientErrorDelegate? = LoggingClientErrorDelegate(),
  473. connectivityStateDelegate: ConnectivityStateDelegate? = nil,
  474. connectivityStateDelegateQueue: DispatchQueue? = nil,
  475. tls: Configuration.TLS? = nil,
  476. connectionBackoff: ConnectionBackoff? = ConnectionBackoff(),
  477. connectionKeepalive: ClientConnectionKeepalive = ClientConnectionKeepalive(),
  478. connectionIdleTimeout: TimeAmount = .minutes(30),
  479. callStartBehavior: CallStartBehavior = .waitsForConnectivity,
  480. httpTargetWindowSize: Int = 8 * 1024 * 1024,
  481. backgroundActivityLogger: Logger = Logger(
  482. label: "io.grpc",
  483. factory: { _ in SwiftLogNoOpLogHandler() }
  484. ),
  485. debugChannelInitializer: (@Sendable (Channel) -> EventLoopFuture<Void>)? = nil
  486. ) {
  487. self.target = target
  488. self.eventLoopGroup = eventLoopGroup
  489. self.errorDelegate = errorDelegate
  490. self.connectivityStateDelegate = connectivityStateDelegate
  491. self.connectivityStateDelegateQueue = connectivityStateDelegateQueue
  492. self.tlsConfiguration = tls.map { GRPCTLSConfiguration(transforming: $0) }
  493. self.connectionBackoff = connectionBackoff
  494. self.connectionKeepalive = connectionKeepalive
  495. self.connectionIdleTimeout = connectionIdleTimeout
  496. self.callStartBehavior = callStartBehavior
  497. self.httpTargetWindowSize = httpTargetWindowSize
  498. self.backgroundActivityLogger = backgroundActivityLogger
  499. self.debugChannelInitializer = debugChannelInitializer
  500. }
  501. #endif // canImport(NIOSSL)
  502. private init(eventLoopGroup: EventLoopGroup, target: ConnectionTarget) {
  503. self.eventLoopGroup = eventLoopGroup
  504. self.target = target
  505. }
  506. /// Make a new configuration using default values.
  507. ///
  508. /// - Parameters:
  509. /// - target: The target to connect to.
  510. /// - eventLoopGroup: The `EventLoopGroup` providing an `EventLoop` for the connection to
  511. /// run on.
  512. /// - Returns: A configuration with default values set.
  513. public static func `default`(
  514. target: ConnectionTarget,
  515. eventLoopGroup: EventLoopGroup
  516. ) -> Configuration {
  517. return .init(eventLoopGroup: eventLoopGroup, target: target)
  518. }
  519. }
  520. }
  521. // MARK: - Configuration helpers/extensions
  522. extension ClientBootstrapProtocol {
  523. /// Connect to the given connection target.
  524. ///
  525. /// - Parameter target: The target to connect to.
  526. func connect(to target: ConnectionTarget) -> EventLoopFuture<Channel> {
  527. switch target.wrapped {
  528. case let .hostAndPort(host, port):
  529. return self.connect(host: host, port: port)
  530. case let .unixDomainSocket(path):
  531. return self.connect(unixDomainSocketPath: path)
  532. case let .socketAddress(address):
  533. return self.connect(to: address)
  534. case let .connectedSocket(socket):
  535. return self.withConnectedSocket(socket)
  536. case let .vsockAddress(address):
  537. return self.connect(to: address)
  538. }
  539. }
  540. }
  541. #if canImport(NIOSSL)
  542. extension ChannelPipeline.SynchronousOperations {
  543. internal func configureNIOSSLForGRPCClient(
  544. sslContext: Result<NIOSSLContext, Error>,
  545. serverHostname: String?,
  546. customVerificationCallback: NIOSSLCustomVerificationCallback?,
  547. logger: Logger
  548. ) throws {
  549. let sslContext = try sslContext.get()
  550. let sslClientHandler: NIOSSLClientHandler
  551. if let customVerificationCallback = customVerificationCallback {
  552. sslClientHandler = try NIOSSLClientHandler(
  553. context: sslContext,
  554. serverHostname: serverHostname,
  555. customVerificationCallback: customVerificationCallback
  556. )
  557. } else {
  558. sslClientHandler = try NIOSSLClientHandler(
  559. context: sslContext,
  560. serverHostname: serverHostname
  561. )
  562. }
  563. try self.addHandler(sslClientHandler)
  564. try self.addHandler(TLSVerificationHandler(logger: logger))
  565. }
  566. }
  567. #endif // canImport(NIOSSL)
  568. extension ChannelPipeline.SynchronousOperations {
  569. internal func configureHTTP2AndGRPCHandlersForGRPCClient(
  570. channel: Channel,
  571. connectionManager: ConnectionManager,
  572. connectionKeepalive: ClientConnectionKeepalive,
  573. connectionIdleTimeout: TimeAmount,
  574. connectionMaxAge: TimeAmount?,
  575. httpTargetWindowSize: Int,
  576. httpMaxFrameSize: Int,
  577. httpMaxResetStreams: Int,
  578. errorDelegate: ClientErrorDelegate?,
  579. logger: Logger
  580. ) throws {
  581. var configuration = NIOHTTP2Handler.ConnectionConfiguration()
  582. configuration.initialSettings = [
  583. // As per the default settings for swift-nio-http2:
  584. HTTP2Setting(parameter: .maxHeaderListSize, value: HPACKDecoder.defaultMaxHeaderListSize),
  585. // We never expect (or allow) server initiated streams.
  586. HTTP2Setting(parameter: .maxConcurrentStreams, value: 0),
  587. // As configured by the user.
  588. HTTP2Setting(parameter: .maxFrameSize, value: httpMaxFrameSize),
  589. HTTP2Setting(parameter: .initialWindowSize, value: httpTargetWindowSize),
  590. ]
  591. configuration.maximumRecentlyResetStreams = httpMaxResetStreams
  592. // We could use 'configureHTTP2Pipeline' here, but we need to add a few handlers between the
  593. // two HTTP/2 handlers so we'll do it manually instead.
  594. try self.addHandler(NIOHTTP2Handler(mode: .client, connectionConfiguration: configuration))
  595. let h2Multiplexer = HTTP2StreamMultiplexer(
  596. mode: .client,
  597. channel: channel,
  598. targetWindowSize: httpTargetWindowSize,
  599. inboundStreamInitializer: nil
  600. )
  601. // The multiplexer is passed through the idle handler so it is only reported on
  602. // successful channel activation - with happy eyeballs multiple pipelines can
  603. // be constructed so it's not safe to report just yet.
  604. try self.addHandler(
  605. GRPCIdleHandler(
  606. connectionManager: connectionManager,
  607. multiplexer: h2Multiplexer,
  608. idleTimeout: connectionIdleTimeout,
  609. maxAge: connectionMaxAge,
  610. keepalive: connectionKeepalive,
  611. logger: logger
  612. )
  613. )
  614. try self.addHandler(h2Multiplexer)
  615. try self.addHandler(DelegatingErrorHandler(logger: logger, delegate: errorDelegate))
  616. }
  617. }
  618. extension Channel {
  619. func configureGRPCClient(
  620. errorDelegate: ClientErrorDelegate?,
  621. logger: Logger
  622. ) -> EventLoopFuture<Void> {
  623. return self.configureHTTP2Pipeline(mode: .client, inboundStreamInitializer: nil).flatMap { _ in
  624. self.pipeline.addHandler(DelegatingErrorHandler(logger: logger, delegate: errorDelegate))
  625. }
  626. }
  627. }
  628. extension TimeAmount {
  629. /// Creates a new `TimeAmount` from the given time interval in seconds.
  630. ///
  631. /// - Parameter timeInterval: The amount of time in seconds
  632. static func seconds(timeInterval: TimeInterval) -> TimeAmount {
  633. return .nanoseconds(Int64(timeInterval * 1_000_000_000))
  634. }
  635. }
  636. extension String {
  637. var isIPAddress: Bool {
  638. // We need some scratch space to let inet_pton write into.
  639. var ipv4Addr = in_addr()
  640. var ipv6Addr = in6_addr()
  641. return self.withCString { ptr in
  642. inet_pton(AF_INET, ptr, &ipv4Addr) == 1 || inet_pton(AF_INET6, ptr, &ipv6Addr) == 1
  643. }
  644. }
  645. }