13
13
//===----------------------------------------------------------------------===//
14
14
15
15
import Foundation
16
+ import Logging
16
17
import NIO
17
18
import NIOConcurrencyHelpers
18
19
import NIOHTTP1
@@ -64,21 +65,29 @@ final class ConnectionPool {
64
65
/// When the pool is asked for a new connection, it creates a `Key` from the url associated to the `request`. This key
65
66
/// is used to determine if there already exists an associated `HTTP1ConnectionProvider` in `providers`.
66
67
/// If there is, the connection provider then takes care of leasing a new connection. If a connection provider doesn't exist, it is created.
67
- func getConnection( for request: HTTPClient . Request , preference: HTTPClient . EventLoopPreference , on eventLoop: EventLoop , deadline: NIODeadline ? , setupComplete: EventLoopFuture < Void > ) -> EventLoopFuture < Connection > {
68
+ func getConnection( _ request: HTTPClient . Request ,
69
+ preference: HTTPClient . EventLoopPreference ,
70
+ taskEventLoop: EventLoop ,
71
+ deadline: NIODeadline ? ,
72
+ setupComplete: EventLoopFuture < Void > ,
73
+ logger: Logger ) -> EventLoopFuture < Connection > {
68
74
let key = Key ( request)
69
75
70
76
let provider : HTTP1ConnectionProvider = self . lock. withLock {
71
77
if let existing = self . providers [ key] , existing. enqueue ( ) {
72
78
return existing
73
79
} else {
74
80
// Connection provider will be created with `pending = 1`
75
- let provider = HTTP1ConnectionProvider ( key: key, eventLoop: eventLoop, configuration: self . configuration, pool: self )
81
+ let provider = HTTP1ConnectionProvider ( key: key,
82
+ eventLoop: taskEventLoop,
83
+ configuration: self . configuration,
84
+ pool: self )
76
85
self . providers [ key] = provider
77
86
return provider
78
87
}
79
88
}
80
89
81
- return provider. getConnection ( preference: preference, setupComplete: setupComplete)
90
+ return provider. getConnection ( preference: preference, setupComplete: setupComplete, logger : logger )
82
91
}
83
92
84
93
func delete( _ provider: HTTP1ConnectionProvider ) {
@@ -167,21 +176,21 @@ class Connection {
167
176
/// Release this `Connection` to its associated `HTTP1ConnectionProvider`.
168
177
///
169
178
/// - Warning: This only releases the connection and doesn't take care of cleaning handlers in the `Channel` pipeline.
170
- func release( closing: Bool ) {
179
+ func release( closing: Bool , logger : Logger ) {
171
180
assert ( self . channel. eventLoop. inEventLoop)
172
- self . provider. release ( connection: self , closing: closing)
181
+ self . provider. release ( connection: self , closing: closing, logger : logger )
173
182
}
174
183
175
184
/// Called when channel exceeds idle time in pool.
176
- func timeout( ) {
185
+ func timeout( logger : Logger ) {
177
186
assert ( self . channel. eventLoop. inEventLoop)
178
- self . provider. timeout ( connection: self )
187
+ self . provider. timeout ( connection: self , logger : logger )
179
188
}
180
189
181
190
/// Called when channel goes inactive while in the pool.
182
- func remoteClosed( ) {
191
+ func remoteClosed( logger : Logger ) {
183
192
assert ( self . channel. eventLoop. inEventLoop)
184
- self . provider. remoteClosed ( connection: self )
193
+ self . provider. remoteClosed ( connection: self , logger : logger )
185
194
}
186
195
187
196
func cancel( ) -> EventLoopFuture < Void > {
@@ -194,9 +203,10 @@ class Connection {
194
203
}
195
204
196
205
/// Sets idle timeout handler and channel inactivity listener.
197
- func setIdleTimeout( timeout: TimeAmount ? ) {
206
+ func setIdleTimeout( timeout: TimeAmount ? , logger : Logger ) {
198
207
_ = self . channel. pipeline. addHandler ( IdleStateHandler ( writeTimeout: timeout) , position: . first) . flatMap { _ in
199
- self . channel. pipeline. addHandler ( IdlePoolConnectionHandler ( connection: self ) )
208
+ self . channel. pipeline. addHandler ( IdlePoolConnectionHandler ( connection: self ,
209
+ logger: logger) )
200
210
}
201
211
}
202
212
@@ -280,43 +290,48 @@ class HTTP1ConnectionProvider {
280
290
self . state. assertInvariants ( )
281
291
}
282
292
283
- private func execute( _ action: Action ) {
293
+ private func execute( _ action: Action , logger : Logger ) {
284
294
switch action {
285
295
case . lease( let connection, let waiter) :
286
296
// if connection is became inactive, we create a new one.
287
297
connection. cancelIdleTimeout ( ) . whenComplete { _ in
288
298
if connection. isActiveEstimation {
299
+ logger. debug ( " reusing active connection \( connection) " )
289
300
waiter. promise. succeed ( connection)
290
301
} else {
302
+ logger. debug ( " opening fresh connection (found \( connection) but we assume it seems inactive) " )
291
303
self . makeChannel ( preference: waiter. preference) . whenComplete { result in
292
- self . connect ( result, waiter: waiter, replacing: connection)
304
+ self . connect ( result, waiter: waiter, replacing: connection, logger : logger )
293
305
}
294
306
}
295
307
}
296
308
case . create( let waiter) :
309
+ logger. debug ( " opening fresh connection (no connections to reuse available) " )
297
310
self . makeChannel ( preference: waiter. preference) . whenComplete { result in
298
- self . connect ( result, waiter: waiter)
311
+ self . connect ( result, waiter: waiter, logger : logger )
299
312
}
300
313
case . replace( let connection, let waiter) :
301
314
connection. cancelIdleTimeout ( ) . flatMap {
302
315
connection. close ( )
303
316
} . whenComplete { _ in
304
317
self . makeChannel ( preference: waiter. preference) . whenComplete { result in
305
- self . connect ( result, waiter: waiter, replacing: connection)
318
+ self . connect ( result, waiter: waiter, replacing: connection, logger : logger )
306
319
}
307
320
}
308
321
case . park( let connection) :
309
- connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool)
322
+ connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool,
323
+ logger: logger. detachingRequestInformation ( nil ) )
310
324
case . closeProvider:
311
325
self . closeAndDelete ( )
312
326
case . none:
313
327
break
314
328
case . parkAnd( let connection, let action) :
315
- connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool)
316
- self . execute ( action)
329
+ connection. setIdleTimeout ( timeout: self . configuration. maximumAllowedIdleTimeInConnectionPool,
330
+ logger: logger. detachingRequestInformation ( nil ) )
331
+ self . execute ( action, logger: logger)
317
332
case . closeAnd( let connection, let action) :
318
333
connection. channel. close ( promise: nil )
319
- self . execute ( action)
334
+ self . execute ( action, logger : logger )
320
335
case . cancel( let connection, let close) :
321
336
connection. cancel ( ) . whenComplete { _ in
322
337
if close {
@@ -335,19 +350,25 @@ class HTTP1ConnectionProvider {
335
350
}
336
351
}
337
352
338
- func getConnection( preference: HTTPClient . EventLoopPreference , setupComplete: EventLoopFuture < Void > ) -> EventLoopFuture < Connection > {
353
+ func getConnection( preference: HTTPClient . EventLoopPreference ,
354
+ setupComplete: EventLoopFuture < Void > ,
355
+ logger: Logger ) -> EventLoopFuture < Connection > {
339
356
let waiter = Waiter ( promise: self . eventLoop. makePromise ( ) , setupComplete: setupComplete, preference: preference)
340
357
341
358
let action : Action = self . lock. withLock {
342
359
self . state. acquire ( waiter: waiter)
343
360
}
344
361
345
- self . execute ( action)
362
+ logger. trace ( " getting a connection with waiter= \( waiter) , action= \( action) " )
363
+ self . execute ( action, logger: logger)
346
364
347
365
return waiter. promise. futureResult
348
366
}
349
367
350
- func connect( _ result: Result < Channel , Error > , waiter: Waiter , replacing closedConnection: Connection ? = nil ) {
368
+ func connect( _ result: Result < Channel , Error > ,
369
+ waiter: Waiter ,
370
+ replacing closedConnection: Connection ? = nil ,
371
+ logger: Logger ) {
351
372
let action : Action
352
373
switch result {
353
374
case . success( let channel) :
@@ -366,11 +387,11 @@ class HTTP1ConnectionProvider {
366
387
waiter. promise. fail ( error)
367
388
}
368
389
waiter. setupComplete. whenComplete { _ in
369
- self . execute ( action)
390
+ self . execute ( action, logger : logger )
370
391
}
371
392
}
372
393
373
- func release( connection: Connection , closing: Bool ) {
394
+ func release( connection: Connection , closing: Bool , logger : Logger ) {
374
395
let action : Action = self . lock. withLock {
375
396
self . state. release ( connection: connection, closing: closing)
376
397
}
@@ -381,31 +402,31 @@ class HTTP1ConnectionProvider {
381
402
case . park, . closeProvider:
382
403
// Since both `.park` and `.deleteProvider` are terminal in terms of execution,
383
404
// we can execute them immediately
384
- self . execute ( action)
405
+ self . execute ( action, logger : logger )
385
406
case . cancel, . closeAnd, . create, . fail, . lease, . parkAnd, . replace:
386
407
// This is needed to start a new stack, otherwise, since this is called on a previous
387
408
// future completion handler chain, it will be growing indefinitely until the connection is closed.
388
409
// We might revisit this when https://github.com/apple/swift-nio/issues/970 is resolved.
389
410
connection. channel. eventLoop. execute {
390
- self . execute ( action)
411
+ self . execute ( action, logger : logger )
391
412
}
392
413
}
393
414
}
394
415
395
- func remoteClosed( connection: Connection ) {
416
+ func remoteClosed( connection: Connection , logger : Logger ) {
396
417
let action : Action = self . lock. withLock {
397
418
self . state. remoteClosed ( connection: connection)
398
419
}
399
420
400
- self . execute ( action)
421
+ self . execute ( action, logger : logger )
401
422
}
402
423
403
- func timeout( connection: Connection ) {
424
+ func timeout( connection: Connection , logger : Logger ) {
404
425
let action : Action = self . lock. withLock {
405
426
self . state. timeout ( connection: connection)
406
427
}
407
428
408
- self . execute ( action)
429
+ self . execute ( action, logger : logger )
409
430
}
410
431
411
432
private func closeAndDelete( ) {
@@ -510,25 +531,27 @@ class IdlePoolConnectionHandler: ChannelInboundHandler, RemovableChannelHandler
510
531
511
532
let connection : Connection
512
533
var eventSent : Bool
534
+ let logger : Logger
513
535
514
- init ( connection: Connection ) {
536
+ init ( connection: Connection , logger : Logger ) {
515
537
self . connection = connection
516
538
self . eventSent = false
539
+ self . logger = logger
517
540
}
518
541
519
542
// this is needed to detect when remote end closes connection while connection is in the pool idling
520
543
func channelInactive( context: ChannelHandlerContext ) {
521
544
if !self . eventSent {
522
545
self . eventSent = true
523
- self . connection. remoteClosed ( )
546
+ self . connection. remoteClosed ( logger : self . logger )
524
547
}
525
548
}
526
549
527
550
func userInboundEventTriggered( context: ChannelHandlerContext , event: Any ) {
528
551
if let idleEvent = event as? IdleStateHandler . IdleStateEvent , idleEvent == . write {
529
552
if !self . eventSent {
530
553
self . eventSent = true
531
- self . connection. timeout ( )
554
+ self . connection. timeout ( logger : self . logger )
532
555
}
533
556
} else {
534
557
context. fireUserInboundEventTriggered ( event)
0 commit comments