Final implementation switch

Author: fabianfett

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -68,7 +68,7 @@ public class HTTPClient{
 publicleteventLoopGroup:EventLoopGroup
 leteventLoopGroupProvider:EventLoopGroupProvider
 letconfiguration:Configuration
-letpool:ConnectionPool
+letpoolManager:HTTPConnectionPool.Manager
 varstate:State
 privateletstateLock=Lock()
 
@@ -110,14 +110,18 @@ public class HTTPClient{
 #endif
 }
 self.configuration = configuration
-self.pool =ConnectionPool(configuration: configuration,
- backgroundActivityLogger: backgroundActivityLogger)
+self.poolManager =HTTPConnectionPool.Manager(
+ eventLoopGroup:self.eventLoopGroup,
+ configuration:self.configuration,
+ backgroundActivityLogger: backgroundActivityLogger
+)
 self.state =.upAndRunning
 }
 
 deinit{
-assert(self.pool.count ==0)
-assert(self.state ==.shutDown,"Client not shut down before the deinit. Please call client.syncShutdown() when no longer needed.")
+guard case .shutDown =self.state else{
+preconditionFailure("Client not shut down before the deinit. Please call client.syncShutdown() when no longer needed.")
+}
 }
 
  /// Shuts down the client and `EventLoopGroup` if it was created by the client.
@@ -175,14 +179,16 @@ public class HTTPClient{
 switchself.eventLoopGroupProvider {
 case.shared:
 self.state =.shutDown
-callback(nil)
+ queue.async{
+callback(nil)
+}
 case.createNew:
 switchself.state {
 case.shuttingDown:
 self.state =.shutDown
 self.eventLoopGroup.shutdownGracefully(queue: queue, callback)
 case.shutDown,.upAndRunning:
-assertionFailure("The only valid state at this point is \(State.shutDown)")
+assertionFailure("The only valid state at this point is \(String(describing:State.shuttingDown))")
 }
 }
 }
@@ -191,33 +197,35 @@ public class HTTPClient{
 privatefunc shutdown(requiresCleanClose:Bool, queue:DispatchQueue, _ callback:@escaping(Error?)->Void){
 do{
 tryself.stateLock.withLock{
-ifself.state !=.upAndRunning{
+guard case .upAndRunning =self.state else{
 throwHTTPClientError.alreadyShutdown
 }
-self.state =.shuttingDown
+self.state =.shuttingDown(requiresCleanClose: requiresCleanClose, callback: callback)
 }
 }catch{
 callback(error)
 return
 }
 
-self.pool.close(on:self.eventLoopGroup.next()).whenComplete{ result in
-varcloseError:Error?
+letpromise=self.eventLoopGroup.next().makePromise(of:Bool.self)
+self.poolManager.shutdown(promise: promise)
+ promise.futureResult.whenComplete{ result in
 switch result {
-case.failure(let error):
- closeError = error
-case.success(let cleanShutdown):
-if !cleanShutdown, requiresCleanClose {
- closeError =HTTPClientError.uncleanShutdown
+case.failure:
+preconditionFailure("Shutting down the connection pool must not fail, ever.")
+case.success(let unclean):
+let(callback, uncleanError)=self.stateLock.withLock{()->((Error?)->Void,Error?)in
+guard case .shuttingDown(let requiresClean, callback:let callback)=self.state else{
+preconditionFailure("Why did the pool manager shut down, if it was not instructed to")
+}
+
+leterror:Error?=(requiresClean && unclean)?HTTPClientError.uncleanShutdown :nil
+return(callback, error)
 }
 
-self.shutdownEventLoop(queue: queue){ eventLoopError in
- // we prioritise .uncleanShutdown here
-iflet error = closeError {
-callback(error)
-}else{
-callback(eventLoopError)
-}
+self.shutdownEventLoop(queue: queue){ error in
+letreportedError= error ?? uncleanError
+callback(reportedError)
 }
 }
 }
@@ -492,7 +500,7 @@ public class HTTPClient{
 lettaskEL:EventLoop
 switch eventLoopPreference.preference {
 case.indifferent:
- taskEL =self.pool.associatedEventLoop(for:ConnectionPool.Key(request))??self.eventLoopGroup.next()
+ taskEL =self.eventLoopGroup.next()
 case.delegate(on:let eventLoop):
 precondition(self.eventLoopGroup.makeIterator().contains{ $0 === eventLoop },"Provided EventLoop must be part of clients EventLoopGroup.")
  taskEL = eventLoop
@@ -540,75 +548,31 @@ public class HTTPClient{
 }
 
 lettask=Task<Delegate.Response>(eventLoop: taskEL, logger: logger)
-letsetupComplete= taskEL.makePromise(of:Void.self)
-letconnection=self.pool.getConnection(request,
- preference: eventLoopPreference,
- taskEventLoop: taskEL,
- deadline: deadline,
- setupComplete: setupComplete.futureResult,
- logger: logger)
-
-lettaskHandler=TaskHandler(task: task,
- kind: request.kind,
- delegate: delegate,
- redirectHandler: redirectHandler,
- ignoreUncleanSSLShutdown:self.configuration.ignoreUncleanSSLShutdown,
- logger: logger)
-
- connection.flatMap{ connection ->EventLoopFuture<Void>in
- logger.debug("got connection for request",
- metadata:["ahc-connection":"\(connection)",
-"ahc-request":"\(request.method)\(request.url)",
-"ahc-channel-el":"\(connection.channel.eventLoop)",
-"ahc-task-el":"\(taskEL)"])
-
-letchannel= connection.channel
-
-func prepareChannelForTask0()->EventLoopFuture<Void>{
-do{
-letsyncPipelineOperations= channel.pipeline.syncOperations
-
-iflet timeout =self.resolve(timeout:self.configuration.timeout.read, deadline: deadline){
-try syncPipelineOperations.addHandler(IdleStateHandler(readTimeout: timeout))
-}
-
-try syncPipelineOperations.addHandler(taskHandler)
-}catch{
- connection.release(closing:true, logger: logger)
-return channel.eventLoop.makeFailedFuture(error)
-}
-
- task.setConnection(connection)
+do{
+letrequestBag=tryRequestBag(
+ request: request,
+ eventLoopPreference: eventLoopPreference,
+ task: task,
+ redirectHandler: redirectHandler,
+ connectionDeadline:.now()+(self.configuration.timeout.connect ??.seconds(10)),
+ idleReadTimeout:self.configuration.timeout.read,
+ delegate: delegate
+)
 
-letisCancelled= task.lock.withLock{
- task.cancelled
+vardeadlineSchedule:Scheduled<Void>?
+iflet deadline = deadline {
+ deadlineSchedule = taskEL.scheduleTask(deadline: deadline){
+ requestBag.fail(HTTPClientError.deadlineExceeded)
 }
 
-if !isCancelled {
-return channel.writeAndFlush(request).flatMapError{ _ in
- // At this point the `TaskHandler` will already be present
- // to handle the failure and pass it to the `promise`
- channel.eventLoop.makeSucceededVoidFuture()
-}
-}else{
-return channel.eventLoop.makeSucceededVoidFuture()
+ task.promise.futureResult.whenComplete{ _ in
+ deadlineSchedule?.cancel()
 }
 }
 
-if channel.eventLoop.inEventLoop {
-returnprepareChannelForTask0()
-}else{
-return channel.eventLoop.flatSubmit{
-returnprepareChannelForTask0()
-}
-}
-}.always{ _ in
- setupComplete.succeed(())
-}.whenFailure{ error in
- taskHandler.callOutToDelegateFireAndForget{ task in
- delegate.didReceiveError(task: task, error)
-}
- task.promise.fail(error)
+self.poolManager.executeRequest(requestBag)
+}catch{
+ task.fail(with: error, delegateType:Delegate.self)
 }
 
 return task
@@ -821,7 +785,7 @@ public class HTTPClient{
 
 enumState{
 case upAndRunning
-case shuttingDown
+case shuttingDown(requiresCleanClose:Bool, callback:(Error?)->Void)
 case shutDown
 }
 }
@@ -926,6 +890,7 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible{
 case serverOfferedUnsupportedApplicationProtocol(String)
 case requestStreamCancelled
 case getConnectionFromPoolTimeout
+case deadlineExceeded
 }
 
 privatevarcode:Code
@@ -995,6 +960,9 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible{
 returnHTTPClientError(code:.serverOfferedUnsupportedApplicationProtocol(proto))
 }
 
+ /// The request deadline was exceeded. The request was cancelled because of this.
+publicstaticletdeadlineExceeded=HTTPClientError(code:.deadlineExceeded)
+
  /// The remote server responded with a status code >= 300, before the full request was sent. The request stream
  /// was therefore cancelled
 publicstaticletrequestStreamCancelled=HTTPClientError(code:.requestStreamCancelled)

Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -658,18 +658,28 @@ extension HTTPClient{
 publicleteventLoop:EventLoop
 
 letpromise:EventLoopPromise<Response>
-varcompletion:EventLoopFuture<Void>
-varconnection:Connection?
-varcancelled:Bool
-letlock:Lock
 letlogger:Logger // We are okay to store the logger here because a Task is for only one request.
 
+varisCancelled:Bool{
+self.lock.withLock{self._isCancelled }
+}
+
+vartaskDelegate:HTTPClientTaskDelegate?{
+get{
+self.lock.withLock{self._taskDelegate }
+}
+set{
+self.lock.withLock{self._taskDelegate = newValue }
+}
+}
+
+privatevar_isCancelled:Bool=false
+privatevar_taskDelegate:HTTPClientTaskDelegate?
+privateletlock=Lock()
+
 init(eventLoop:EventLoop, logger:Logger){
 self.eventLoop = eventLoop
 self.promise = eventLoop.makePromise()
-self.completion =self.promise.futureResult.map{ _ in}
-self.cancelled =false
-self.lock =Lock()
 self.logger = logger
 }
 
@@ -694,69 +704,24 @@ extension HTTPClient{
 
  /// Cancels the request execution.
 publicfunc cancel(){
-letchannel:Channel?=self.lock.withLock{
-if !self.cancelled {
-self.cancelled =true
-returnself.connection?.channel
-}else{
-returnnil
-}
+lettaskDelegate=self.lock.withLock{()->HTTPClientTaskDelegate?in
+self._isCancelled =true
+returnself._taskDelegate
 }
- channel?.triggerUserOutboundEvent(TaskCancelEvent(), promise:nil)
-}
 
-@discardableResult
-func setConnection(_ connection:Connection)->Connection{
-returnself.lock.withLock{
-self.connection = connection
-ifself.cancelled {
- connection.channel.triggerUserOutboundEvent(TaskCancelEvent(), promise:nil)
-}
-return connection
-}
+ taskDelegate?.cancel()
 }
 
 func succeed<Delegate:HTTPClientResponseDelegate>(promise:EventLoopPromise<Response>?,
  with value:Response,
  delegateType:Delegate.Type,
  closing:Bool){
-self.releaseAssociatedConnection(delegateType: delegateType,
- closing: closing).whenSuccess{
- promise?.succeed(value)
-}
+ promise?.succeed(value)
 }
 
 func fail<Delegate:HTTPClientResponseDelegate>(with error:Error,
  delegateType:Delegate.Type){
-iflet connection =self.connection {
-self.releaseAssociatedConnection(delegateType: delegateType, closing:true)
-.whenSuccess{
-self.promise.fail(error)
- connection.channel.close(promise:nil)
-}
-}else{
- // this is used in tests where we don't want to bootstrap the whole connection pool
-self.promise.fail(error)
-}
-}
-
-func releaseAssociatedConnection<Delegate:HTTPClientResponseDelegate>(delegateType:Delegate.Type,
- closing:Bool)->EventLoopFuture<Void>{
-iflet connection =self.connection {
- // remove read timeout handler
-return connection.removeHandler(IdleStateHandler.self).flatMap{
- connection.removeHandler(TaskHandler<Delegate>.self)
-}.map{
- connection.release(closing: closing, logger:self.logger)
-}.flatMapError{ error in
-fatalError("Couldn't remove taskHandler: \(error)")
-}
-}else{
- // TODO: This seems only reached in some internal unit test
- // Maybe there could be a better handling in the future to make
- // it an error outside of testing contexts
-returnself.eventLoop.makeSucceededFuture(())
-}
+self.promise.fail(error)
 }
 }
 }
@@ -1076,9 +1041,7 @@ extension TaskHandler: ChannelDuplexHandler{
 break
 case.redirected(let head,let redirectURL):
 self.state =.endOrError
-self.task.releaseAssociatedConnection(delegateType:Delegate.self, closing:self.closing).whenSuccess{
-self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise:self.task.promise)
-}
+self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise:self.task.promise)
 default:
 self.state =.bufferedEnd
 self.handleReadForDelegate(response, context: context)

Sources/AsyncHTTPClient/RequestBag.swift

@@ -66,10 +66,10 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>{
 self.requestFramingMetadata = metadata
 
  // TODO: comment in once we switch to using the Request bag in AHC
-// self.task.taskDelegate = self
-// self.task.futureResult.whenComplete{_ in
-// self.task.taskDelegate = nil
-// }
+self.task.taskDelegate =self
+self.task.futureResult.whenComplete{ _ in
+self.task.taskDelegate =nil
+}
 }
 
 privatefunc requestWasQueued0(_ scheduler:HTTPRequestScheduler){
@@ -111,7 +111,7 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>{
 self.writeNextRequestPart($0)
 }
 
- body.stream(writer).whenComplete{
+ body.stream(writer).hop(to:self.eventLoop).whenComplete{
 self.finishRequestBodyStream($0)
 }
 
@@ -140,7 +140,7 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>{
 }
 
 privatefunc writeNextRequestPart0(_ part:IOData)->EventLoopFuture<Void>{
-self.task.eventLoop.assertInEventLoop()
+self.eventLoop.assertInEventLoop()
 
 letaction=self.state.writeNextRequestPart(part, taskEventLoop:self.task.eventLoop)