Ensure HTTPConnectionPool calls shutdown on active connections. (#422)

Trigger the correct method on HTTPConnectionPool.Connection to cancel the request and close the connection.

Author: fabianfett

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

@@ -235,7 +235,7 @@ final class HTTPConnectionPool{
 }
 
 forconnectionin cleanupContext.cancel {
- connection.close(promise:nil)
+ connection.shutdown()
 }
 
 forconnectionIDin cleanupContext.connectBackoff {

Tests/AsyncHTTPClientTests/HTTPConnectionPoolTests+XCTest.swift

@@ -31,6 +31,7 @@ extension HTTPConnectionPoolTests{
 ("testConnectionCreationIsRetriedUntilRequestIsFailed", testConnectionCreationIsRetriedUntilRequestIsFailed),
 ("testConnectionCreationIsRetriedUntilPoolIsShutdown", testConnectionCreationIsRetriedUntilPoolIsShutdown),
 ("testConnectionCreationIsRetriedUntilRequestIsCancelled", testConnectionCreationIsRetriedUntilRequestIsCancelled),
+("testConnectionShutdownIsCalledOnActiveConnections", testConnectionShutdownIsCalledOnActiveConnections),
 ]
 }
 }

Tests/AsyncHTTPClientTests/HTTPConnectionPoolTests.swift

@@ -294,7 +294,7 @@ class HTTPConnectionPoolTests: XCTestCase{
 leteventLoop= eventLoopGroup.next()
 defer{XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully())}
 
-letrequest=try!HTTPClient.Request(url:"http://localhost:9000")
+letrequest=try!HTTPClient.Request(url:"http://localhost:\(httpBin.port)")
 letpoolDelegate=TestDelegate(eventLoop: eventLoop)
 
 letpool=HTTPConnectionPool(
@@ -318,7 +318,7 @@ class HTTPConnectionPoolTests: XCTestCase{
 
 varmaybeRequest:HTTPClient.Request?
 varmaybeRequestBag:RequestBag<ResponseAccumulator>?
-XCTAssertNoThrow(maybeRequest =tryHTTPClient.Request(url:"https://localhost:\(httpBin.port)"))
+XCTAssertNoThrow(maybeRequest =tryHTTPClient.Request(url:"http://localhost:\(httpBin.port)"))
 XCTAssertNoThrow(maybeRequestBag =tryRequestBag(
  request:XCTUnwrap(maybeRequest),
  eventLoopPreference:.indifferent,
@@ -332,13 +332,63 @@ class HTTPConnectionPoolTests: XCTestCase{
 guardlet requestBag = maybeRequestBag else{returnXCTFail("Expected to get a request")}
 
  pool.executeRequest(requestBag)
-XCTAssertNoThrow(try eventLoop.scheduleTask(in:.milliseconds(100)){}.futureResult.wait())
+XCTAssertNoThrow(try eventLoop.scheduleTask(in:.seconds(1)){}.futureResult.wait())
  requestBag.cancel()
 
 XCTAssertThrowsError(try requestBag.task.futureResult.wait()){
 XCTAssertEqual($0 as?HTTPClientError,.cancelled)
 }
+XCTAssertGreaterThanOrEqual(httpBin.createdConnections,3)
+}
+
+func testConnectionShutdownIsCalledOnActiveConnections(){
+lethttpBin=HTTPBin()
+defer{XCTAssertNoThrow(try httpBin.shutdown())}
+leteventLoopGroup=MultiThreadedEventLoopGroup(numberOfThreads:1)
+leteventLoop= eventLoopGroup.next()
+defer{XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully())}
+
+letrequest=try!HTTPClient.Request(url:"http://localhost:\(httpBin.port)")
+letpoolDelegate=TestDelegate(eventLoop: eventLoop)
+
+letpool=HTTPConnectionPool(
+ eventLoopGroup: eventLoopGroup,
+ sslContextCache:.init(),
+ tlsConfiguration:.none,
+ clientConfiguration:.init(),
+ key:.init(request),
+ delegate: poolDelegate,
+ idGenerator:.init(),
+ backgroundActivityLogger:.init(label:"test")
+)
+
+varmaybeRequest:HTTPClient.Request?
+varmaybeRequestBag:RequestBag<ResponseAccumulator>?
+XCTAssertNoThrow(maybeRequest =tryHTTPClient.Request(url:"http://localhost:\(httpBin.port)/wait"))
+XCTAssertNoThrow(maybeRequestBag =tryRequestBag(
+ request:XCTUnwrap(maybeRequest),
+ eventLoopPreference:.indifferent,
+ task:.init(eventLoop: eventLoopGroup.next(), logger:.init(label:"test")),
+ redirectHandler:nil,
+ connectionDeadline:.now()+.seconds(5),
+ idleReadTimeout:nil,
+ delegate:ResponseAccumulator(request:XCTUnwrap(maybeRequest))
+))
+
+guardlet requestBag = maybeRequestBag else{returnXCTFail("Expected to get a request")}
+
+ pool.executeRequest(requestBag)
+XCTAssertNoThrow(try eventLoop.scheduleTask(in:.milliseconds(500)){}.futureResult.wait())
+ pool.shutdown()
+
+XCTAssertNoThrow(try poolDelegate.future.wait())
+
+XCTAssertThrowsError(try requestBag.task.futureResult.wait()){
+XCTAssertEqual($0 as?HTTPClientError,.cancelled)
+}
+
 XCTAssertGreaterThanOrEqual(httpBin.createdConnections,1)
+XCTAssertGreaterThanOrEqual(httpBin.activeConnections,0)
 }
 }