ClientConnection.swift 27 KB

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