ClientConnection.swift 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754
  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 Foundation
  17. import Logging
  18. import NIO
  19. import NIOHTTP2
  20. import NIOSSL
  21. import NIOTLS
  22. import NIOTransportServices
  23. import SwiftProtobuf
  24. /// Provides a single, managed connection to a server.
  25. ///
  26. /// The connection to the server is provided by a single channel which will attempt to reconnect
  27. /// to the server if the connection is dropped. This connection is guaranteed to always use the same
  28. /// event loop.
  29. ///
  30. /// The connection is initially setup with a handler to verify that TLS was established
  31. /// successfully (assuming TLS is being used).
  32. ///
  33. /// ┌──────────────────────────┐
  34. /// │ DelegatingErrorHandler │
  35. /// └──────────▲───────────────┘
  36. /// HTTP2Frame│
  37. /// ┌──────────┴───────────────┐
  38. /// │ SettingsObservingHandler │
  39. /// └──────────▲───────────────┘
  40. /// HTTP2Frame│
  41. /// │ ⠇ ⠇ ⠇ ⠇
  42. /// │ ┌┴─▼┐ ┌┴─▼┐
  43. /// │ │ | │ | HTTP/2 streams
  44. /// │ └▲─┬┘ └▲─┬┘
  45. /// │ │ │ │ │ HTTP2Frame
  46. /// ┌─┴────────────────┴─▼───┴─▼┐
  47. /// │ HTTP2StreamMultiplexer |
  48. /// └─▲───────────────────────┬─┘
  49. /// HTTP2Frame│ │HTTP2Frame
  50. /// ┌─┴───────────────────────▼─┐
  51. /// │ NIOHTTP2Handler │
  52. /// └─▲───────────────────────┬─┘
  53. /// ByteBuffer│ │ByteBuffer
  54. /// ┌─┴───────────────────────▼─┐
  55. /// │ TLSVerificationHandler │
  56. /// └─▲───────────────────────┬─┘
  57. /// ByteBuffer│ │ByteBuffer
  58. /// ┌─┴───────────────────────▼─┐
  59. /// │ NIOSSLHandler │
  60. /// └─▲───────────────────────┬─┘
  61. /// ByteBuffer│ │ByteBuffer
  62. /// │ ▼
  63. ///
  64. /// The `TLSVerificationHandler` observes the outcome of the SSL handshake and determines
  65. /// whether a `ClientConnection` should be returned to the user. In either eventuality, the
  66. /// handler removes itself from the pipeline once TLS has been verified. There is also a handler
  67. /// after the multiplexer for observing the initial settings frame, after which it determines that
  68. /// the connection state is `.ready` and removes itself from the channel. Finally there is a
  69. /// delegated error handler which uses the error delegate associated with this connection
  70. /// (see `DelegatingErrorHandler`).
  71. ///
  72. /// See `BaseClientCall` for a description of the pipelines associated with each HTTP/2 stream.
  73. public class ClientConnection {
  74. private let connectionManager: ConnectionManager
  75. private func getChannel() -> EventLoopFuture<Channel> {
  76. switch self.configuration.callStartBehavior.wrapped {
  77. case .waitsForConnectivity:
  78. return self.connectionManager.getChannel()
  79. case .fastFailure:
  80. return self.connectionManager.getOptimisticChannel()
  81. }
  82. }
  83. /// HTTP multiplexer from the `channel` handling gRPC calls.
  84. internal var multiplexer: EventLoopFuture<HTTP2StreamMultiplexer> {
  85. return self.getChannel().flatMap {
  86. $0.pipeline.handler(type: HTTP2StreamMultiplexer.self)
  87. }
  88. }
  89. /// The configuration for this client.
  90. internal let configuration: Configuration
  91. internal let scheme: String
  92. internal let authority: String
  93. /// A monitor for the connectivity state.
  94. public var connectivity: ConnectivityStateMonitor {
  95. return self.connectionManager.monitor
  96. }
  97. /// The `EventLoop` this connection is using.
  98. public var eventLoop: EventLoop {
  99. return self.connectionManager.eventLoop
  100. }
  101. /// Creates a new connection from the given configuration. Prefer using
  102. /// `ClientConnection.secure(group:)` to build a connection secured with TLS or
  103. /// `ClientConnection.insecure(group:)` to build a plaintext connection.
  104. ///
  105. /// - Important: Users should prefer using `ClientConnection.secure(group:)` to build a connection
  106. /// with TLS, or `ClientConnection.insecure(group:)` to build a connection without TLS.
  107. public init(configuration: Configuration) {
  108. self.configuration = configuration
  109. self.scheme = configuration.tls == nil ? "http" : "https"
  110. self.authority = configuration.target.host
  111. self.connectionManager = ConnectionManager(
  112. configuration: configuration,
  113. logger: configuration.backgroundActivityLogger
  114. )
  115. }
  116. /// Closes the connection to the server.
  117. public func close() -> EventLoopFuture<Void> {
  118. return self.connectionManager.shutdown()
  119. }
  120. /// Extracts a logger and request ID from the call options and returns them. The logger will
  121. /// be populated with the request ID (if applicable) and any metadata from the connection manager.
  122. private func populatedLoggerAndRequestID(from callOptions: CallOptions) -> (Logger, String?) {
  123. var logger = callOptions.logger
  124. self.connectionManager.appendMetadata(to: &logger)
  125. // Attach the request ID.
  126. let requestID = callOptions.requestIDProvider.requestID()
  127. if let requestID = requestID {
  128. logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
  129. }
  130. return (logger, requestID)
  131. }
  132. private func makeRequestHead(path: String, options: CallOptions,
  133. requestID: String?) -> _GRPCRequestHead {
  134. return _GRPCRequestHead(
  135. scheme: self.scheme,
  136. path: path,
  137. host: self.authority,
  138. options: options,
  139. requestID: requestID
  140. )
  141. }
  142. }
  143. // MARK: - Unary
  144. extension ClientConnection: GRPCChannel {
  145. private func makeUnaryCall<Serializer: MessageSerializer, Deserializer: MessageDeserializer>(
  146. serializer: Serializer,
  147. deserializer: Deserializer,
  148. path: String,
  149. request: Serializer.Input,
  150. callOptions: CallOptions
  151. ) -> UnaryCall<Serializer.Input, Deserializer.Output> {
  152. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  153. logger.debug("starting rpc", metadata: ["path": "\(path)"])
  154. let call = UnaryCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  155. multiplexer: self.multiplexer,
  156. serializer: serializer,
  157. deserializer: deserializer,
  158. callOptions: callOptions,
  159. errorDelegate: self.configuration.errorDelegate,
  160. logger: logger
  161. )
  162. call.send(
  163. self.makeRequestHead(path: path, options: callOptions, requestID: requestID),
  164. request: request
  165. )
  166. return call
  167. }
  168. /// A unary call using `SwiftProtobuf.Message` messages.
  169. public func makeUnaryCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
  170. path: String,
  171. request: Request,
  172. callOptions: CallOptions
  173. ) -> UnaryCall<Request, Response> {
  174. return self.makeUnaryCall(
  175. serializer: ProtobufSerializer(),
  176. deserializer: ProtobufDeserializer(),
  177. path: path,
  178. request: request,
  179. callOptions: callOptions
  180. )
  181. }
  182. /// A unary call using `GRPCPayload` messages.
  183. public func makeUnaryCall<Request: GRPCPayload, Response: GRPCPayload>(
  184. path: String,
  185. request: Request,
  186. callOptions: CallOptions
  187. ) -> UnaryCall<Request, Response> {
  188. return self.makeUnaryCall(
  189. serializer: GRPCPayloadSerializer(),
  190. deserializer: GRPCPayloadDeserializer(),
  191. path: path,
  192. request: request,
  193. callOptions: callOptions
  194. )
  195. }
  196. }
  197. // MARK: - Client Streaming
  198. extension ClientConnection {
  199. private func makeClientStreamingCall<
  200. Serializer: MessageSerializer,
  201. Deserializer: MessageDeserializer
  202. >(
  203. serializer: Serializer,
  204. deserializer: Deserializer,
  205. path: String,
  206. callOptions: CallOptions
  207. ) -> ClientStreamingCall<Serializer.Input, Deserializer.Output> {
  208. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  209. logger.debug("starting rpc", metadata: ["path": "\(path)"])
  210. let call = ClientStreamingCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  211. multiplexer: self.multiplexer,
  212. serializer: serializer,
  213. deserializer: deserializer,
  214. callOptions: callOptions,
  215. errorDelegate: self.configuration.errorDelegate,
  216. logger: logger
  217. )
  218. call.sendHead(self.makeRequestHead(path: path, options: callOptions, requestID: requestID))
  219. return call
  220. }
  221. /// A client streaming call using `SwiftProtobuf.Message` messages.
  222. public func makeClientStreamingCall<
  223. Request: SwiftProtobuf.Message,
  224. Response: SwiftProtobuf.Message
  225. >(
  226. path: String,
  227. callOptions: CallOptions
  228. ) -> ClientStreamingCall<Request, Response> {
  229. return self.makeClientStreamingCall(
  230. serializer: ProtobufSerializer(),
  231. deserializer: ProtobufDeserializer(),
  232. path: path,
  233. callOptions: callOptions
  234. )
  235. }
  236. /// A client streaming call using `GRPCPayload` messages.
  237. public func makeClientStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  238. path: String,
  239. callOptions: CallOptions
  240. ) -> ClientStreamingCall<Request, Response> {
  241. return self.makeClientStreamingCall(
  242. serializer: GRPCPayloadSerializer(),
  243. deserializer: GRPCPayloadDeserializer(),
  244. path: path,
  245. callOptions: callOptions
  246. )
  247. }
  248. }
  249. // MARK: - Server Streaming
  250. extension ClientConnection {
  251. private func makeServerStreamingCall<
  252. Serializer: MessageSerializer,
  253. Deserializer: MessageDeserializer
  254. >(
  255. serializer: Serializer,
  256. deserializer: Deserializer,
  257. path: String,
  258. request: Serializer.Input,
  259. callOptions: CallOptions,
  260. handler: @escaping (Deserializer.Output) -> Void
  261. ) -> ServerStreamingCall<Serializer.Input, Deserializer.Output> {
  262. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  263. logger.debug("starting rpc", metadata: ["path": "\(path)"])
  264. let call = ServerStreamingCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  265. multiplexer: self.multiplexer,
  266. serializer: serializer,
  267. deserializer: deserializer,
  268. callOptions: callOptions,
  269. errorDelegate: self.configuration.errorDelegate,
  270. logger: logger,
  271. responseHandler: handler
  272. )
  273. call.send(
  274. self.makeRequestHead(path: path, options: callOptions, requestID: requestID),
  275. request: request
  276. )
  277. return call
  278. }
  279. /// A server streaming call using `SwiftProtobuf.Message` messages.
  280. public func makeServerStreamingCall<
  281. Request: SwiftProtobuf.Message,
  282. Response: SwiftProtobuf.Message
  283. >(
  284. path: String,
  285. request: Request,
  286. callOptions: CallOptions,
  287. handler: @escaping (Response) -> Void
  288. ) -> ServerStreamingCall<Request, Response> {
  289. return self.makeServerStreamingCall(
  290. serializer: ProtobufSerializer(),
  291. deserializer: ProtobufDeserializer(),
  292. path: path,
  293. request: request,
  294. callOptions: callOptions,
  295. handler: handler
  296. )
  297. }
  298. /// A server streaming call using `GRPCPayload` messages.
  299. public func makeServerStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  300. path: String,
  301. request: Request,
  302. callOptions: CallOptions,
  303. handler: @escaping (Response) -> Void
  304. ) -> ServerStreamingCall<Request, Response> {
  305. return self.makeServerStreamingCall(
  306. serializer: GRPCPayloadSerializer(),
  307. deserializer: GRPCPayloadDeserializer(),
  308. path: path,
  309. request: request,
  310. callOptions: callOptions,
  311. handler: handler
  312. )
  313. }
  314. }
  315. // MARK: - Bidirectional Streaming
  316. extension ClientConnection {
  317. private func makeBidirectionalStreamingCall<
  318. Serializer: MessageSerializer,
  319. Deserializer: MessageDeserializer
  320. >(
  321. serializer: Serializer,
  322. deserializer: Deserializer,
  323. path: String,
  324. callOptions: CallOptions,
  325. handler: @escaping (Deserializer.Output) -> Void
  326. ) -> BidirectionalStreamingCall<Serializer.Input, Deserializer.Output> {
  327. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  328. logger.debug("starting rpc", metadata: ["path": "\(path)"])
  329. let call = BidirectionalStreamingCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  330. multiplexer: self.multiplexer,
  331. serializer: serializer,
  332. deserializer: deserializer,
  333. callOptions: callOptions,
  334. errorDelegate: self.configuration.errorDelegate,
  335. logger: logger,
  336. responseHandler: handler
  337. )
  338. call.sendHead(self.makeRequestHead(path: path, options: callOptions, requestID: requestID))
  339. return call
  340. }
  341. /// A bidirectional streaming call using `SwiftProtobuf.Message` messages.
  342. public func makeBidirectionalStreamingCall<
  343. Request: SwiftProtobuf.Message,
  344. Response: SwiftProtobuf.Message
  345. >(
  346. path: String,
  347. callOptions: CallOptions,
  348. handler: @escaping (Response) -> Void
  349. ) -> BidirectionalStreamingCall<Request, Response> {
  350. return self.makeBidirectionalStreamingCall(
  351. serializer: ProtobufSerializer(),
  352. deserializer: ProtobufDeserializer(),
  353. path: path,
  354. callOptions: callOptions,
  355. handler: handler
  356. )
  357. }
  358. /// A bidirectional streaming call using `GRPCPayload` messages.
  359. public func makeBidirectionalStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  360. path: String,
  361. callOptions: CallOptions,
  362. handler: @escaping (Response) -> Void
  363. ) -> BidirectionalStreamingCall<Request, Response> {
  364. return self.makeBidirectionalStreamingCall(
  365. serializer: GRPCPayloadSerializer(),
  366. deserializer: GRPCPayloadDeserializer(),
  367. path: path,
  368. callOptions: callOptions,
  369. handler: handler
  370. )
  371. }
  372. }
  373. // MARK: - Configuration structures
  374. /// A target to connect to.
  375. public struct ConnectionTarget {
  376. internal enum Wrapped {
  377. case hostAndPort(String, Int)
  378. case unixDomainSocket(String)
  379. case socketAddress(SocketAddress)
  380. }
  381. internal var wrapped: Wrapped
  382. private init(_ wrapped: Wrapped) {
  383. self.wrapped = wrapped
  384. }
  385. /// The host and port.
  386. public static func hostAndPort(_ host: String, _ port: Int) -> ConnectionTarget {
  387. return ConnectionTarget(.hostAndPort(host, port))
  388. }
  389. /// The path of a Unix domain socket.
  390. public static func unixDomainSocket(_ path: String) -> ConnectionTarget {
  391. return ConnectionTarget(.unixDomainSocket(path))
  392. }
  393. /// A NIO socket address.
  394. public static func socketAddress(_ address: SocketAddress) -> ConnectionTarget {
  395. return ConnectionTarget(.socketAddress(address))
  396. }
  397. var host: String {
  398. switch self.wrapped {
  399. case let .hostAndPort(host, _):
  400. return host
  401. case let .socketAddress(.v4(address)):
  402. return address.host
  403. case let .socketAddress(.v6(address)):
  404. return address.host
  405. case .unixDomainSocket, .socketAddress(.unixDomainSocket):
  406. return "localhost"
  407. }
  408. }
  409. }
  410. /// The connectivity behavior to use when starting an RPC.
  411. public struct CallStartBehavior: Hashable {
  412. internal enum Behavior: Hashable {
  413. case waitsForConnectivity
  414. case fastFailure
  415. }
  416. internal var wrapped: Behavior
  417. private init(_ wrapped: Behavior) {
  418. self.wrapped = wrapped
  419. }
  420. /// Waits for connectivity (that is, the 'ready' connectivity state) before attempting to start
  421. /// an RPC. Doing so may involve multiple connection attempts.
  422. ///
  423. /// This is the preferred, and default, behaviour.
  424. public static let waitsForConnectivity = CallStartBehavior(.waitsForConnectivity)
  425. /// The 'fast failure' behaviour is intended for cases where users would rather their RPC failed
  426. /// quickly rather than waiting for an active connection. The behaviour depends on the current
  427. /// connectivity state:
  428. ///
  429. /// - Idle: a connection attempt will be started and the RPC will fail if that attempt fails.
  430. /// - Connecting: a connection attempt is already in progress, the RPC will fail if that attempt
  431. /// fails.
  432. /// - Ready: a connection is already active: the RPC will be started using that connection.
  433. /// - Transient failure: the last connection or connection attempt failed and gRPC is waiting to
  434. /// connect again. The RPC will fail immediately.
  435. /// - Shutdown: the connection is shutdown, the RPC will fail immediately.
  436. public static let fastFailure = CallStartBehavior(.fastFailure)
  437. }
  438. extension ClientConnection {
  439. /// The configuration for a connection.
  440. public struct Configuration {
  441. /// The target to connect to.
  442. public var target: ConnectionTarget
  443. /// The event loop group to run the connection on.
  444. public var eventLoopGroup: EventLoopGroup
  445. /// An error delegate which is called when errors are caught. Provided delegates **must not
  446. /// maintain a strong reference to this `ClientConnection`**. Doing so will cause a retain
  447. /// cycle.
  448. public var errorDelegate: ClientErrorDelegate?
  449. /// A delegate which is called when the connectivity state is changed.
  450. public var connectivityStateDelegate: ConnectivityStateDelegate?
  451. /// The `DispatchQueue` on which to call the connectivity state delegate. If a delegate is
  452. /// provided but the queue is `nil` then one will be created by gRPC.
  453. public var connectivityStateDelegateQueue: DispatchQueue?
  454. /// TLS configuration for this connection. `nil` if TLS is not desired.
  455. public var tls: TLS?
  456. /// The connection backoff configuration. If no connection retrying is required then this should
  457. /// be `nil`.
  458. public var connectionBackoff: ConnectionBackoff?
  459. /// The connection keepalive configuration.
  460. public var connectionKeepalive: ClientConnectionKeepalive
  461. /// The amount of time to wait before closing the connection. The idle timeout will start only
  462. /// if there are no RPCs in progress and will be cancelled as soon as any RPCs start.
  463. ///
  464. /// If a connection becomes idle, starting a new RPC will automatically create a new connection.
  465. public var connectionIdleTimeout: TimeAmount
  466. /// The behavior used to determine when an RPC should start. That is, whether it should wait for
  467. /// an active connection or fail quickly if no connection is currently available.
  468. public var callStartBehavior: CallStartBehavior
  469. /// The HTTP/2 flow control target window size.
  470. public var httpTargetWindowSize: Int
  471. /// The HTTP protocol used for this connection.
  472. public var httpProtocol: HTTP2FramePayloadToHTTP1ClientCodec.HTTPProtocol {
  473. return self.tls == nil ? .http : .https
  474. }
  475. /// A logger for background information (such as connectivity state). A separate logger for
  476. /// requests may be provided in the `CallOptions`.
  477. ///
  478. /// Defaults to a no-op logger.
  479. public var backgroundActivityLogger: Logger
  480. /// A channel initializer which will be run after gRPC has initialized each channel. This may be
  481. /// used to add additional handlers to the pipeline and is intended for debugging.
  482. ///
  483. /// - Warning: The initializer closure may be invoked *multiple times*.
  484. public var debugChannelInitializer: ((Channel) -> EventLoopFuture<Void>)?
  485. /// Create a `Configuration` with some pre-defined defaults. Prefer using
  486. /// `ClientConnection.secure(group:)` to build a connection secured with TLS or
  487. /// `ClientConnection.insecure(group:)` to build a plaintext connection.
  488. ///
  489. /// - Parameter target: The target to connect to.
  490. /// - Parameter eventLoopGroup: The event loop group to run the connection on.
  491. /// - Parameter errorDelegate: The error delegate, defaulting to a delegate which will log only
  492. /// on debug builds.
  493. /// - Parameter connectivityStateDelegate: A connectivity state delegate, defaulting to `nil`.
  494. /// - Parameter connectivityStateDelegateQueue: A `DispatchQueue` on which to call the
  495. /// `connectivityStateDelegate`.
  496. /// - Parameter tls: TLS configuration, defaulting to `nil`.
  497. /// - Parameter connectionBackoff: The connection backoff configuration to use.
  498. /// - Parameter connectionKeepalive: The keepalive configuration to use.
  499. /// - Parameter connectionIdleTimeout: The amount of time to wait before closing the connection, defaulting to 5 minutes.
  500. /// - Parameter callStartBehavior: The behavior used to determine when a call should start in
  501. /// relation to its underlying connection. Defaults to `waitsForConnectivity`.
  502. /// - Parameter httpTargetWindowSize: The HTTP/2 flow control target window size.
  503. /// - Parameter backgroundActivityLogger: A logger for background information (such as
  504. /// connectivity state). Defaults to a no-op logger.
  505. /// - Parameter debugChannelInitializer: A channel initializer will be called after gRPC has
  506. /// initialized the channel. Defaults to `nil`.
  507. public init(
  508. target: ConnectionTarget,
  509. eventLoopGroup: EventLoopGroup,
  510. errorDelegate: ClientErrorDelegate? = LoggingClientErrorDelegate(),
  511. connectivityStateDelegate: ConnectivityStateDelegate? = nil,
  512. connectivityStateDelegateQueue: DispatchQueue? = nil,
  513. tls: Configuration.TLS? = nil,
  514. connectionBackoff: ConnectionBackoff? = ConnectionBackoff(),
  515. connectionKeepalive: ClientConnectionKeepalive = ClientConnectionKeepalive(),
  516. connectionIdleTimeout: TimeAmount = .minutes(5),
  517. callStartBehavior: CallStartBehavior = .waitsForConnectivity,
  518. httpTargetWindowSize: Int = 65535,
  519. backgroundActivityLogger: Logger = Logger(
  520. label: "io.grpc",
  521. factory: { _ in SwiftLogNoOpLogHandler() }
  522. ),
  523. debugChannelInitializer: ((Channel) -> EventLoopFuture<Void>)? = nil
  524. ) {
  525. self.target = target
  526. self.eventLoopGroup = eventLoopGroup
  527. self.errorDelegate = errorDelegate
  528. self.connectivityStateDelegate = connectivityStateDelegate
  529. self.connectivityStateDelegateQueue = connectivityStateDelegateQueue
  530. self.tls = tls
  531. self.connectionBackoff = connectionBackoff
  532. self.connectionKeepalive = connectionKeepalive
  533. self.connectionIdleTimeout = connectionIdleTimeout
  534. self.callStartBehavior = callStartBehavior
  535. self.httpTargetWindowSize = httpTargetWindowSize
  536. self.backgroundActivityLogger = backgroundActivityLogger
  537. self.debugChannelInitializer = debugChannelInitializer
  538. }
  539. }
  540. }
  541. // MARK: - Configuration helpers/extensions
  542. extension ClientBootstrapProtocol {
  543. /// Connect to the given connection target.
  544. ///
  545. /// - Parameter target: The target to connect to.
  546. func connect(to target: ConnectionTarget) -> EventLoopFuture<Channel> {
  547. switch target.wrapped {
  548. case let .hostAndPort(host, port):
  549. return self.connect(host: host, port: port)
  550. case let .unixDomainSocket(path):
  551. return self.connect(unixDomainSocketPath: path)
  552. case let .socketAddress(address):
  553. return self.connect(to: address)
  554. }
  555. }
  556. }
  557. extension Channel {
  558. /// Configure the channel with TLS.
  559. ///
  560. /// This function adds two handlers to the pipeline: the `NIOSSLClientHandler` to handle TLS, and
  561. /// the `TLSVerificationHandler` which verifies that a successful handshake was completed.
  562. ///
  563. /// - Parameter configuration: The configuration to configure the channel with.
  564. /// - Parameter serverHostname: The server hostname to use if the hostname should be verified.
  565. /// - Parameter errorDelegate: The error delegate to use for the TLS verification handler.
  566. func configureTLS(
  567. _ configuration: TLSConfiguration,
  568. serverHostname: String?,
  569. errorDelegate: ClientErrorDelegate?,
  570. logger: Logger
  571. ) -> EventLoopFuture<Void> {
  572. do {
  573. let sslClientHandler = try NIOSSLClientHandler(
  574. context: try NIOSSLContext(configuration: configuration),
  575. serverHostname: serverHostname
  576. )
  577. return self.pipeline.addHandlers(sslClientHandler, TLSVerificationHandler(logger: logger))
  578. } catch {
  579. return self.eventLoop.makeFailedFuture(error)
  580. }
  581. }
  582. func configureGRPCClient(
  583. httpTargetWindowSize: Int,
  584. tlsConfiguration: TLSConfiguration?,
  585. tlsServerHostname: String?,
  586. connectionManager: ConnectionManager,
  587. connectionKeepalive: ClientConnectionKeepalive,
  588. connectionIdleTimeout: TimeAmount,
  589. errorDelegate: ClientErrorDelegate?,
  590. requiresZeroLengthWriteWorkaround: Bool,
  591. logger: Logger
  592. ) -> EventLoopFuture<Void> {
  593. let tlsConfigured = tlsConfiguration.map {
  594. self.configureTLS(
  595. $0,
  596. serverHostname: tlsServerHostname,
  597. errorDelegate: errorDelegate,
  598. logger: logger
  599. )
  600. }
  601. let configuration: EventLoopFuture<Void> = (
  602. tlsConfigured ?? self.eventLoop
  603. .makeSucceededFuture(())
  604. ).flatMap {
  605. self.configureHTTP2Pipeline(
  606. mode: .client,
  607. targetWindowSize: httpTargetWindowSize,
  608. inboundStreamInitializer: nil
  609. )
  610. }.flatMap { _ in
  611. self.pipeline.handler(type: NIOHTTP2Handler.self).flatMap { http2Handler in
  612. self.pipeline.addHandlers(
  613. [
  614. GRPCClientKeepaliveHandler(configuration: connectionKeepalive),
  615. GRPCIdleHandler(mode: .client(connectionManager), idleTimeout: connectionIdleTimeout),
  616. ],
  617. position: .after(http2Handler)
  618. )
  619. }.flatMap {
  620. let errorHandler = DelegatingErrorHandler(
  621. logger: logger,
  622. delegate: errorDelegate
  623. )
  624. return self.pipeline.addHandler(errorHandler)
  625. }
  626. }
  627. #if canImport(Network)
  628. // This availability guard is arguably unnecessary, but we add it anyway.
  629. if requiresZeroLengthWriteWorkaround, #available(
  630. OSX 10.14,
  631. iOS 12.0,
  632. tvOS 12.0,
  633. watchOS 6.0,
  634. *
  635. ) {
  636. return configuration.flatMap {
  637. self.pipeline.addHandler(NIOFilterEmptyWritesHandler(), position: .first)
  638. }
  639. } else {
  640. return configuration
  641. }
  642. #else
  643. return configuration
  644. #endif
  645. }
  646. func configureGRPCClient(
  647. errorDelegate: ClientErrorDelegate?,
  648. logger: Logger
  649. ) -> EventLoopFuture<Void> {
  650. return self.configureHTTP2Pipeline(mode: .client, inboundStreamInitializer: nil).flatMap { _ in
  651. self.pipeline.addHandler(DelegatingErrorHandler(logger: logger, delegate: errorDelegate))
  652. }
  653. }
  654. }
  655. extension TimeAmount {
  656. /// Creates a new `TimeAmount` from the given time interval in seconds.
  657. ///
  658. /// - Parameter timeInterval: The amount of time in seconds
  659. static func seconds(timeInterval: TimeInterval) -> TimeAmount {
  660. return .nanoseconds(Int64(timeInterval * 1_000_000_000))
  661. }
  662. }
  663. extension String {
  664. var isIPAddress: Bool {
  665. // We need some scratch space to let inet_pton write into.
  666. var ipv4Addr = in_addr()
  667. var ipv6Addr = in6_addr()
  668. return self.withCString { ptr in
  669. inet_pton(AF_INET, ptr, &ipv4Addr) == 1 ||
  670. inet_pton(AF_INET6, ptr, &ipv6Addr) == 1
  671. }
  672. }
  673. }