ClientConnection.swift 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822
  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.tls?.hostnameOverride ?? 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. /// Populates the logger in `options` and appends a request ID header to the metadata, if
  133. /// configured.
  134. /// - Parameter options: The options containing the logger to populate.
  135. private func populateLogger(in options: inout CallOptions) {
  136. // Get connection metadata.
  137. self.connectionManager.appendMetadata(to: &options.logger)
  138. // Attach a request ID.
  139. let requestID = options.requestIDProvider.requestID()
  140. if let requestID = requestID {
  141. options.logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
  142. // Add the request ID header too.
  143. if let requestIDHeader = options.requestIDHeader {
  144. options.customMetadata.add(name: requestIDHeader, value: requestID)
  145. }
  146. }
  147. }
  148. private func makeRequestHead(path: String, options: CallOptions,
  149. requestID: String?) -> _GRPCRequestHead {
  150. return _GRPCRequestHead(
  151. scheme: self.scheme,
  152. path: path,
  153. host: self.authority,
  154. options: options,
  155. requestID: requestID
  156. )
  157. }
  158. }
  159. extension ClientConnection {
  160. public func makeCall<Request: Message, Response: Message>(
  161. path: String,
  162. type: GRPCCallType,
  163. callOptions: CallOptions,
  164. interceptors: [ClientInterceptor<Request, Response>]
  165. ) -> Call<Request, Response> {
  166. var options = callOptions
  167. self.populateLogger(in: &options)
  168. return Call(
  169. path: path,
  170. type: type,
  171. eventLoop: self.multiplexer.eventLoop,
  172. options: options,
  173. interceptors: interceptors,
  174. transportFactory: .http2(
  175. multiplexer: self.multiplexer,
  176. authority: self.authority,
  177. scheme: self.scheme,
  178. errorDelegate: self.configuration.errorDelegate
  179. )
  180. )
  181. }
  182. public func makeCall<Request: GRPCPayload, Response: GRPCPayload>(
  183. path: String,
  184. type: GRPCCallType,
  185. callOptions: CallOptions,
  186. interceptors: [ClientInterceptor<Request, Response>]
  187. ) -> Call<Request, Response> {
  188. var options = callOptions
  189. self.populateLogger(in: &options)
  190. return Call(
  191. path: path,
  192. type: type,
  193. eventLoop: self.multiplexer.eventLoop,
  194. options: options,
  195. interceptors: interceptors,
  196. transportFactory: .http2(
  197. multiplexer: self.multiplexer,
  198. authority: self.authority,
  199. scheme: self.scheme,
  200. errorDelegate: self.configuration.errorDelegate
  201. )
  202. )
  203. }
  204. }
  205. // MARK: - Unary
  206. extension ClientConnection: GRPCChannel {
  207. private func makeUnaryCall<Serializer: MessageSerializer, Deserializer: MessageDeserializer>(
  208. serializer: Serializer,
  209. deserializer: Deserializer,
  210. path: String,
  211. request: Serializer.Input,
  212. callOptions: CallOptions
  213. ) -> UnaryCall<Serializer.Input, Deserializer.Output> {
  214. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  215. let call = UnaryCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  216. multiplexer: self.multiplexer,
  217. serializer: serializer,
  218. deserializer: deserializer,
  219. callOptions: callOptions,
  220. errorDelegate: self.configuration.errorDelegate,
  221. logger: logger
  222. )
  223. call.send(
  224. self.makeRequestHead(path: path, options: callOptions, requestID: requestID),
  225. request: request
  226. )
  227. return call
  228. }
  229. /// A unary call using `SwiftProtobuf.Message` messages.
  230. public func makeUnaryCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
  231. path: String,
  232. request: Request,
  233. callOptions: CallOptions
  234. ) -> UnaryCall<Request, Response> {
  235. return self.makeUnaryCall(
  236. serializer: ProtobufSerializer(),
  237. deserializer: ProtobufDeserializer(),
  238. path: path,
  239. request: request,
  240. callOptions: callOptions
  241. )
  242. }
  243. /// A unary call using `GRPCPayload` messages.
  244. public func makeUnaryCall<Request: GRPCPayload, Response: GRPCPayload>(
  245. path: String,
  246. request: Request,
  247. callOptions: CallOptions
  248. ) -> UnaryCall<Request, Response> {
  249. return self.makeUnaryCall(
  250. serializer: GRPCPayloadSerializer(),
  251. deserializer: GRPCPayloadDeserializer(),
  252. path: path,
  253. request: request,
  254. callOptions: callOptions
  255. )
  256. }
  257. }
  258. // MARK: - Client Streaming
  259. extension ClientConnection {
  260. private func makeClientStreamingCall<
  261. Serializer: MessageSerializer,
  262. Deserializer: MessageDeserializer
  263. >(
  264. serializer: Serializer,
  265. deserializer: Deserializer,
  266. path: String,
  267. callOptions: CallOptions
  268. ) -> ClientStreamingCall<Serializer.Input, Deserializer.Output> {
  269. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  270. let call = ClientStreamingCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  271. multiplexer: self.multiplexer,
  272. serializer: serializer,
  273. deserializer: deserializer,
  274. callOptions: callOptions,
  275. errorDelegate: self.configuration.errorDelegate,
  276. logger: logger
  277. )
  278. call.sendHead(self.makeRequestHead(path: path, options: callOptions, requestID: requestID))
  279. return call
  280. }
  281. /// A client streaming call using `SwiftProtobuf.Message` messages.
  282. public func makeClientStreamingCall<
  283. Request: SwiftProtobuf.Message,
  284. Response: SwiftProtobuf.Message
  285. >(
  286. path: String,
  287. callOptions: CallOptions
  288. ) -> ClientStreamingCall<Request, Response> {
  289. return self.makeClientStreamingCall(
  290. serializer: ProtobufSerializer(),
  291. deserializer: ProtobufDeserializer(),
  292. path: path,
  293. callOptions: callOptions
  294. )
  295. }
  296. /// A client streaming call using `GRPCPayload` messages.
  297. public func makeClientStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  298. path: String,
  299. callOptions: CallOptions
  300. ) -> ClientStreamingCall<Request, Response> {
  301. return self.makeClientStreamingCall(
  302. serializer: GRPCPayloadSerializer(),
  303. deserializer: GRPCPayloadDeserializer(),
  304. path: path,
  305. callOptions: callOptions
  306. )
  307. }
  308. }
  309. // MARK: - Server Streaming
  310. extension ClientConnection {
  311. private func makeServerStreamingCall<
  312. Serializer: MessageSerializer,
  313. Deserializer: MessageDeserializer
  314. >(
  315. serializer: Serializer,
  316. deserializer: Deserializer,
  317. path: String,
  318. request: Serializer.Input,
  319. callOptions: CallOptions,
  320. handler: @escaping (Deserializer.Output) -> Void
  321. ) -> ServerStreamingCall<Serializer.Input, Deserializer.Output> {
  322. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  323. let call = ServerStreamingCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  324. multiplexer: self.multiplexer,
  325. serializer: serializer,
  326. deserializer: deserializer,
  327. callOptions: callOptions,
  328. errorDelegate: self.configuration.errorDelegate,
  329. logger: logger,
  330. responseHandler: handler
  331. )
  332. call.send(
  333. self.makeRequestHead(path: path, options: callOptions, requestID: requestID),
  334. request: request
  335. )
  336. return call
  337. }
  338. /// A server streaming call using `SwiftProtobuf.Message` messages.
  339. public func makeServerStreamingCall<
  340. Request: SwiftProtobuf.Message,
  341. Response: SwiftProtobuf.Message
  342. >(
  343. path: String,
  344. request: Request,
  345. callOptions: CallOptions,
  346. handler: @escaping (Response) -> Void
  347. ) -> ServerStreamingCall<Request, Response> {
  348. return self.makeServerStreamingCall(
  349. serializer: ProtobufSerializer(),
  350. deserializer: ProtobufDeserializer(),
  351. path: path,
  352. request: request,
  353. callOptions: callOptions,
  354. handler: handler
  355. )
  356. }
  357. /// A server streaming call using `GRPCPayload` messages.
  358. public func makeServerStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  359. path: String,
  360. request: Request,
  361. callOptions: CallOptions,
  362. handler: @escaping (Response) -> Void
  363. ) -> ServerStreamingCall<Request, Response> {
  364. return self.makeServerStreamingCall(
  365. serializer: GRPCPayloadSerializer(),
  366. deserializer: GRPCPayloadDeserializer(),
  367. path: path,
  368. request: request,
  369. callOptions: callOptions,
  370. handler: handler
  371. )
  372. }
  373. }
  374. // MARK: - Bidirectional Streaming
  375. extension ClientConnection {
  376. private func makeBidirectionalStreamingCall<
  377. Serializer: MessageSerializer,
  378. Deserializer: MessageDeserializer
  379. >(
  380. serializer: Serializer,
  381. deserializer: Deserializer,
  382. path: String,
  383. callOptions: CallOptions,
  384. handler: @escaping (Deserializer.Output) -> Void
  385. ) -> BidirectionalStreamingCall<Serializer.Input, Deserializer.Output> {
  386. let (logger, requestID) = self.populatedLoggerAndRequestID(from: callOptions)
  387. let call = BidirectionalStreamingCall<Serializer.Input, Deserializer.Output>.makeOnHTTP2Stream(
  388. multiplexer: self.multiplexer,
  389. serializer: serializer,
  390. deserializer: deserializer,
  391. callOptions: callOptions,
  392. errorDelegate: self.configuration.errorDelegate,
  393. logger: logger,
  394. responseHandler: handler
  395. )
  396. call.sendHead(self.makeRequestHead(path: path, options: callOptions, requestID: requestID))
  397. return call
  398. }
  399. /// A bidirectional streaming call using `SwiftProtobuf.Message` messages.
  400. public func makeBidirectionalStreamingCall<
  401. Request: SwiftProtobuf.Message,
  402. Response: SwiftProtobuf.Message
  403. >(
  404. path: String,
  405. callOptions: CallOptions,
  406. handler: @escaping (Response) -> Void
  407. ) -> BidirectionalStreamingCall<Request, Response> {
  408. return self.makeBidirectionalStreamingCall(
  409. serializer: ProtobufSerializer(),
  410. deserializer: ProtobufDeserializer(),
  411. path: path,
  412. callOptions: callOptions,
  413. handler: handler
  414. )
  415. }
  416. /// A bidirectional streaming call using `GRPCPayload` messages.
  417. public func makeBidirectionalStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  418. path: String,
  419. callOptions: CallOptions,
  420. handler: @escaping (Response) -> Void
  421. ) -> BidirectionalStreamingCall<Request, Response> {
  422. return self.makeBidirectionalStreamingCall(
  423. serializer: GRPCPayloadSerializer(),
  424. deserializer: GRPCPayloadDeserializer(),
  425. path: path,
  426. callOptions: callOptions,
  427. handler: handler
  428. )
  429. }
  430. }
  431. // MARK: - Configuration structures
  432. /// A target to connect to.
  433. public struct ConnectionTarget {
  434. internal enum Wrapped {
  435. case hostAndPort(String, Int)
  436. case unixDomainSocket(String)
  437. case socketAddress(SocketAddress)
  438. }
  439. internal var wrapped: Wrapped
  440. private init(_ wrapped: Wrapped) {
  441. self.wrapped = wrapped
  442. }
  443. /// The host and port.
  444. public static func hostAndPort(_ host: String, _ port: Int) -> ConnectionTarget {
  445. return ConnectionTarget(.hostAndPort(host, port))
  446. }
  447. /// The path of a Unix domain socket.
  448. public static func unixDomainSocket(_ path: String) -> ConnectionTarget {
  449. return ConnectionTarget(.unixDomainSocket(path))
  450. }
  451. /// A NIO socket address.
  452. public static func socketAddress(_ address: SocketAddress) -> ConnectionTarget {
  453. return ConnectionTarget(.socketAddress(address))
  454. }
  455. var host: String {
  456. switch self.wrapped {
  457. case let .hostAndPort(host, _):
  458. return host
  459. case let .socketAddress(.v4(address)):
  460. return address.host
  461. case let .socketAddress(.v6(address)):
  462. return address.host
  463. case .unixDomainSocket, .socketAddress(.unixDomainSocket):
  464. return "localhost"
  465. }
  466. }
  467. }
  468. /// The connectivity behavior to use when starting an RPC.
  469. public struct CallStartBehavior: Hashable {
  470. internal enum Behavior: Hashable {
  471. case waitsForConnectivity
  472. case fastFailure
  473. }
  474. internal var wrapped: Behavior
  475. private init(_ wrapped: Behavior) {
  476. self.wrapped = wrapped
  477. }
  478. /// Waits for connectivity (that is, the 'ready' connectivity state) before attempting to start
  479. /// an RPC. Doing so may involve multiple connection attempts.
  480. ///
  481. /// This is the preferred, and default, behaviour.
  482. public static let waitsForConnectivity = CallStartBehavior(.waitsForConnectivity)
  483. /// The 'fast failure' behaviour is intended for cases where users would rather their RPC failed
  484. /// quickly rather than waiting for an active connection. The behaviour depends on the current
  485. /// connectivity state:
  486. ///
  487. /// - Idle: a connection attempt will be started and the RPC will fail if that attempt fails.
  488. /// - Connecting: a connection attempt is already in progress, the RPC will fail if that attempt
  489. /// fails.
  490. /// - Ready: a connection is already active: the RPC will be started using that connection.
  491. /// - Transient failure: the last connection or connection attempt failed and gRPC is waiting to
  492. /// connect again. The RPC will fail immediately.
  493. /// - Shutdown: the connection is shutdown, the RPC will fail immediately.
  494. public static let fastFailure = CallStartBehavior(.fastFailure)
  495. }
  496. extension ClientConnection {
  497. /// The configuration for a connection.
  498. public struct Configuration {
  499. /// The target to connect to.
  500. public var target: ConnectionTarget
  501. /// The event loop group to run the connection on.
  502. public var eventLoopGroup: EventLoopGroup
  503. /// An error delegate which is called when errors are caught. Provided delegates **must not
  504. /// maintain a strong reference to this `ClientConnection`**. Doing so will cause a retain
  505. /// cycle.
  506. public var errorDelegate: ClientErrorDelegate?
  507. /// A delegate which is called when the connectivity state is changed.
  508. public var connectivityStateDelegate: ConnectivityStateDelegate?
  509. /// The `DispatchQueue` on which to call the connectivity state delegate. If a delegate is
  510. /// provided but the queue is `nil` then one will be created by gRPC.
  511. public var connectivityStateDelegateQueue: DispatchQueue?
  512. /// TLS configuration for this connection. `nil` if TLS is not desired.
  513. public var tls: TLS?
  514. /// The connection backoff configuration. If no connection retrying is required then this should
  515. /// be `nil`.
  516. public var connectionBackoff: ConnectionBackoff?
  517. /// The connection keepalive configuration.
  518. public var connectionKeepalive: ClientConnectionKeepalive
  519. /// The amount of time to wait before closing the connection. The idle timeout will start only
  520. /// if there are no RPCs in progress and will be cancelled as soon as any RPCs start.
  521. ///
  522. /// If a connection becomes idle, starting a new RPC will automatically create a new connection.
  523. public var connectionIdleTimeout: TimeAmount
  524. /// The behavior used to determine when an RPC should start. That is, whether it should wait for
  525. /// an active connection or fail quickly if no connection is currently available.
  526. public var callStartBehavior: CallStartBehavior
  527. /// The HTTP/2 flow control target window size.
  528. public var httpTargetWindowSize: Int
  529. /// The HTTP protocol used for this connection.
  530. public var httpProtocol: HTTP2FramePayloadToHTTP1ClientCodec.HTTPProtocol {
  531. return self.tls == nil ? .http : .https
  532. }
  533. /// A logger for background information (such as connectivity state). A separate logger for
  534. /// requests may be provided in the `CallOptions`.
  535. ///
  536. /// Defaults to a no-op logger.
  537. public var backgroundActivityLogger: Logger
  538. /// A channel initializer which will be run after gRPC has initialized each channel. This may be
  539. /// used to add additional handlers to the pipeline and is intended for debugging.
  540. ///
  541. /// - Warning: The initializer closure may be invoked *multiple times*.
  542. public var debugChannelInitializer: ((Channel) -> EventLoopFuture<Void>)?
  543. /// Create a `Configuration` with some pre-defined defaults. Prefer using
  544. /// `ClientConnection.secure(group:)` to build a connection secured with TLS or
  545. /// `ClientConnection.insecure(group:)` to build a plaintext connection.
  546. ///
  547. /// - Parameter target: The target to connect to.
  548. /// - Parameter eventLoopGroup: The event loop group to run the connection on.
  549. /// - Parameter errorDelegate: The error delegate, defaulting to a delegate which will log only
  550. /// on debug builds.
  551. /// - Parameter connectivityStateDelegate: A connectivity state delegate, defaulting to `nil`.
  552. /// - Parameter connectivityStateDelegateQueue: A `DispatchQueue` on which to call the
  553. /// `connectivityStateDelegate`.
  554. /// - Parameter tls: TLS configuration, defaulting to `nil`.
  555. /// - Parameter connectionBackoff: The connection backoff configuration to use.
  556. /// - Parameter connectionKeepalive: The keepalive configuration to use.
  557. /// - Parameter connectionIdleTimeout: The amount of time to wait before closing the connection, defaulting to 30 minutes.
  558. /// - Parameter callStartBehavior: The behavior used to determine when a call should start in
  559. /// relation to its underlying connection. Defaults to `waitsForConnectivity`.
  560. /// - Parameter httpTargetWindowSize: The HTTP/2 flow control target window size.
  561. /// - Parameter backgroundActivityLogger: A logger for background information (such as
  562. /// connectivity state). Defaults to a no-op logger.
  563. /// - Parameter debugChannelInitializer: A channel initializer will be called after gRPC has
  564. /// initialized the channel. Defaults to `nil`.
  565. public init(
  566. target: ConnectionTarget,
  567. eventLoopGroup: EventLoopGroup,
  568. errorDelegate: ClientErrorDelegate? = LoggingClientErrorDelegate(),
  569. connectivityStateDelegate: ConnectivityStateDelegate? = nil,
  570. connectivityStateDelegateQueue: DispatchQueue? = nil,
  571. tls: Configuration.TLS? = nil,
  572. connectionBackoff: ConnectionBackoff? = ConnectionBackoff(),
  573. connectionKeepalive: ClientConnectionKeepalive = ClientConnectionKeepalive(),
  574. connectionIdleTimeout: TimeAmount = .minutes(30),
  575. callStartBehavior: CallStartBehavior = .waitsForConnectivity,
  576. httpTargetWindowSize: Int = 65535,
  577. backgroundActivityLogger: Logger = Logger(
  578. label: "io.grpc",
  579. factory: { _ in SwiftLogNoOpLogHandler() }
  580. ),
  581. debugChannelInitializer: ((Channel) -> EventLoopFuture<Void>)? = nil
  582. ) {
  583. self.target = target
  584. self.eventLoopGroup = eventLoopGroup
  585. self.errorDelegate = errorDelegate
  586. self.connectivityStateDelegate = connectivityStateDelegate
  587. self.connectivityStateDelegateQueue = connectivityStateDelegateQueue
  588. self.tls = tls
  589. self.connectionBackoff = connectionBackoff
  590. self.connectionKeepalive = connectionKeepalive
  591. self.connectionIdleTimeout = connectionIdleTimeout
  592. self.callStartBehavior = callStartBehavior
  593. self.httpTargetWindowSize = httpTargetWindowSize
  594. self.backgroundActivityLogger = backgroundActivityLogger
  595. self.debugChannelInitializer = debugChannelInitializer
  596. }
  597. }
  598. }
  599. // MARK: - Configuration helpers/extensions
  600. extension ClientBootstrapProtocol {
  601. /// Connect to the given connection target.
  602. ///
  603. /// - Parameter target: The target to connect to.
  604. func connect(to target: ConnectionTarget) -> EventLoopFuture<Channel> {
  605. switch target.wrapped {
  606. case let .hostAndPort(host, port):
  607. return self.connect(host: host, port: port)
  608. case let .unixDomainSocket(path):
  609. return self.connect(unixDomainSocketPath: path)
  610. case let .socketAddress(address):
  611. return self.connect(to: address)
  612. }
  613. }
  614. }
  615. extension Channel {
  616. /// Configure the channel with TLS.
  617. ///
  618. /// This function adds two handlers to the pipeline: the `NIOSSLClientHandler` to handle TLS, and
  619. /// the `TLSVerificationHandler` which verifies that a successful handshake was completed.
  620. ///
  621. /// - Parameter configuration: The configuration to configure the channel with.
  622. /// - Parameter serverHostname: The server hostname to use if the hostname should be verified.
  623. /// - Parameter errorDelegate: The error delegate to use for the TLS verification handler.
  624. func configureTLS(
  625. _ configuration: TLSConfiguration,
  626. serverHostname: String?,
  627. errorDelegate: ClientErrorDelegate?,
  628. logger: Logger
  629. ) -> EventLoopFuture<Void> {
  630. do {
  631. let sslClientHandler = try NIOSSLClientHandler(
  632. context: try NIOSSLContext(configuration: configuration),
  633. serverHostname: serverHostname
  634. )
  635. return self.pipeline.addHandlers(sslClientHandler, TLSVerificationHandler(logger: logger))
  636. } catch {
  637. return self.eventLoop.makeFailedFuture(error)
  638. }
  639. }
  640. func configureGRPCClient(
  641. httpTargetWindowSize: Int,
  642. tlsConfiguration: TLSConfiguration?,
  643. tlsServerHostname: String?,
  644. connectionManager: ConnectionManager,
  645. connectionKeepalive: ClientConnectionKeepalive,
  646. connectionIdleTimeout: TimeAmount,
  647. errorDelegate: ClientErrorDelegate?,
  648. requiresZeroLengthWriteWorkaround: Bool,
  649. logger: Logger
  650. ) -> EventLoopFuture<Void> {
  651. let tlsConfigured = tlsConfiguration.map {
  652. self.configureTLS(
  653. $0,
  654. serverHostname: tlsServerHostname,
  655. errorDelegate: errorDelegate,
  656. logger: logger
  657. )
  658. }
  659. let configuration: EventLoopFuture<Void> = (
  660. tlsConfigured ?? self.eventLoop
  661. .makeSucceededFuture(())
  662. ).flatMap {
  663. self.configureHTTP2Pipeline(
  664. mode: .client,
  665. targetWindowSize: httpTargetWindowSize,
  666. inboundStreamInitializer: nil
  667. )
  668. }.flatMap { _ in
  669. self.pipeline.handler(type: NIOHTTP2Handler.self).flatMap { http2Handler in
  670. self.pipeline.addHandlers(
  671. [
  672. GRPCClientKeepaliveHandler(configuration: connectionKeepalive),
  673. GRPCIdleHandler(
  674. mode: .client(connectionManager),
  675. logger: logger,
  676. idleTimeout: connectionIdleTimeout
  677. ),
  678. ],
  679. position: .after(http2Handler)
  680. )
  681. }.flatMap {
  682. let errorHandler = DelegatingErrorHandler(
  683. logger: logger,
  684. delegate: errorDelegate
  685. )
  686. return self.pipeline.addHandler(errorHandler)
  687. }
  688. }
  689. #if canImport(Network)
  690. // This availability guard is arguably unnecessary, but we add it anyway.
  691. if requiresZeroLengthWriteWorkaround, #available(
  692. OSX 10.14,
  693. iOS 12.0,
  694. tvOS 12.0,
  695. watchOS 6.0,
  696. *
  697. ) {
  698. return configuration.flatMap {
  699. self.pipeline.addHandler(NIOFilterEmptyWritesHandler(), position: .first)
  700. }
  701. } else {
  702. return configuration
  703. }
  704. #else
  705. return configuration
  706. #endif
  707. }
  708. func configureGRPCClient(
  709. errorDelegate: ClientErrorDelegate?,
  710. logger: Logger
  711. ) -> EventLoopFuture<Void> {
  712. return self.configureHTTP2Pipeline(mode: .client, inboundStreamInitializer: nil).flatMap { _ in
  713. self.pipeline.addHandler(DelegatingErrorHandler(logger: logger, delegate: errorDelegate))
  714. }
  715. }
  716. }
  717. extension TimeAmount {
  718. /// Creates a new `TimeAmount` from the given time interval in seconds.
  719. ///
  720. /// - Parameter timeInterval: The amount of time in seconds
  721. static func seconds(timeInterval: TimeInterval) -> TimeAmount {
  722. return .nanoseconds(Int64(timeInterval * 1_000_000_000))
  723. }
  724. }
  725. extension String {
  726. var isIPAddress: Bool {
  727. // We need some scratch space to let inet_pton write into.
  728. var ipv4Addr = in_addr()
  729. var ipv6Addr = in6_addr()
  730. return self.withCString { ptr in
  731. inet_pton(AF_INET, ptr, &ipv4Addr) == 1 ||
  732. inet_pton(AF_INET6, ptr, &ipv6Addr) == 1
  733. }
  734. }
  735. }