Introduce generic type ConnectionType into ConnectionsState

Motivation:

To rework tests of ConnectionsState we want to have a "simpler" version
of Connection to be used, therefore here we convert ConnectionsState to
support generic type ConnectionType. We will substitute current
Connection with a test version in follow up commit.

Modifications:

ConnectionsState is altered to work on generic type ConnectionType
instead of solid type Connection.

Users of ConnectionsState are modified to provide type Connection into
ConnectionType in this commit.

Result:

Test are passing, no observable behaviour change.

Author: ya-pulser

Sources/AsyncHTTPClient/Connection.swift

@@ -153,14 +153,14 @@ extension Connection: CustomStringConvertible {
     }
 }
 
-struct ConnectionKey: Hashable {
-    let connection: Connection
+struct ConnectionKey<ConnectionType>: Hashable where ConnectionType: PoolManageableConnection {
+    let connection: ConnectionType
 
-    init(_ connection: Connection) {
+    init(_ connection: ConnectionType) {
         self.connection = connection
     }
 
-    static func == (lhs: ConnectionKey, rhs: ConnectionKey) -> Bool {
+    static func == (lhs: ConnectionKey<ConnectionType>, rhs: ConnectionKey<ConnectionType>) -> Bool {
         return ObjectIdentifier(lhs.connection) == ObjectIdentifier(rhs.connection)
     }
 

Sources/AsyncHTTPClient/ConnectionPool.swift

@@ -191,7 +191,7 @@ class HTTP1ConnectionProvider {
 
     var closePromise: EventLoopPromise<Void>
 
-    var state: ConnectionsState
+    var state: ConnectionsState<Connection>
 
     private let backgroundActivityLogger: Logger
 
@@ -222,7 +222,7 @@ class HTTP1ConnectionProvider {
         self.state.assertInvariants()
     }
 
-    func execute(_ action: Action, logger: Logger) {
+    func execute(_ action: Action<Connection>, logger: Logger) {
         switch action {
         case .lease(let connection, let waiter):
             // if connection is became inactive, we create a new one.
@@ -297,7 +297,7 @@ class HTTP1ConnectionProvider {
     func getConnection(preference: HTTPClient.EventLoopPreference,
                        setupComplete: EventLoopFuture<Void>,
                        logger: Logger) -> EventLoopFuture<Connection> {
-        let waiter = Waiter(promise: self.eventLoop.makePromise(), setupComplete: setupComplete, preference: preference)
+        let waiter = Waiter<Connection>(promise: self.eventLoop.makePromise(), setupComplete: setupComplete, preference: preference)
 
         let action: Action = self.lock.withLock {
             self.state.acquire(waiter: waiter)
@@ -309,10 +309,10 @@ class HTTP1ConnectionProvider {
     }
 
     func connect(_ result: Result<Channel, Error>,
-                 waiter: Waiter,
+                 waiter: Waiter<Connection>,
                  replacing closedConnection: Connection? = nil,
                  logger: Logger) {
-        let action: Action
+        let action: Action<Connection>
         switch result {
         case .success(let channel):
             logger.trace("successfully created connection",
@@ -478,9 +478,9 @@ class HTTP1ConnectionProvider {
     ///
     /// `Waiter`s are created when `maximumConcurrentConnections` is reached
     /// and we cannot create new connections anymore.
-    struct Waiter {
+    struct Waiter<ConnectionType: PoolManageableConnection> {
         /// The promise to complete once a connection is available
-        let promise: EventLoopPromise<Connection>
+        let promise: EventLoopPromise<ConnectionType>
 
         /// Future that will be succeeded when request timeout handler and `TaskHandler` are added to the pipeline.
         let setupComplete: EventLoopFuture<Void>

Sources/AsyncHTTPClient/ConnectionsState.swift

@@ -15,29 +15,29 @@
 import NIO
 
 extension HTTP1ConnectionProvider {
-    enum Action {
-        case lease(Connection, Waiter)
-        case create(Waiter)
-        case replace(Connection, Waiter)
+    enum Action<ConnectionType: PoolManageableConnection> {
+        case lease(ConnectionType, Waiter<ConnectionType>)
+        case create(Waiter<ConnectionType>)
+        case replace(ConnectionType, Waiter<ConnectionType>)
         case closeProvider
-        case park(Connection)
+        case park(ConnectionType)
         case none
-        case fail(Waiter, Error)
-        indirect case closeAnd(Connection, Action)
-        indirect case parkAnd(Connection, Action)
+        case fail(Waiter<ConnectionType>, Error)
+        indirect case closeAnd(ConnectionType, Action<ConnectionType>)
+        indirect case parkAnd(ConnectionType, Action<ConnectionType>)
     }
 
-    struct ConnectionsState {
+    struct ConnectionsState<ConnectionType: PoolManageableConnection> {
         enum State {
             case active
             case closed
         }
 
-        struct Snapshot {
+        struct Snapshot<ConnectionType: PoolManageableConnection> {
             var state: State
-            var availableConnections: CircularBuffer<Connection>
-            var leasedConnections: Set<ConnectionKey>
-            var waiters: CircularBuffer<Waiter>
+            var availableConnections: CircularBuffer<ConnectionType>
+            var leasedConnections: Set<ConnectionKey<ConnectionType>>
+            var waiters: CircularBuffer<Waiter<ConnectionType>>
             var openedConnectionsCount: Int
             var pending: Int
         }
@@ -48,16 +48,16 @@ extension HTTP1ConnectionProvider {
         private var state: State = .active
 
         /// Opened connections that are available.
-        private var availableConnections: CircularBuffer<Connection> = .init(initialCapacity: 8)
+        private var availableConnections: CircularBuffer<ConnectionType> = .init(initialCapacity: 8)
 
         /// Opened connections that are leased to the user.
-        private var leasedConnections: Set<ConnectionKey> = .init()
+        private var leasedConnections: Set<ConnectionKey<ConnectionType>> = .init()
 
         /// Consumers that weren't able to get a new connection without exceeding
         /// `maximumConcurrentConnections` get a `Future<Connection>`
         /// whose associated promise is stored in `Waiter`. The promise is completed
         /// as soon as possible by the provider, in FIFO order.
-        private var waiters: CircularBuffer<Waiter> = .init(initialCapacity: 8)
+        private var waiters: CircularBuffer<Waiter<ConnectionType>> = .init(initialCapacity: 8)
 
         /// Number of opened or opening connections, used to keep track of all connections and enforcing `maximumConcurrentConnections` limit.
         private var openedConnectionsCount: Int = 0
@@ -70,11 +70,11 @@ extension HTTP1ConnectionProvider {
             self.eventLoop = eventLoop
         }
 
-        func testsOnly_getInternalState() -> Snapshot {
+        func testsOnly_getInternalState() -> Snapshot<ConnectionType> {
             return Snapshot(state: self.state, availableConnections: self.availableConnections, leasedConnections: self.leasedConnections, waiters: self.waiters, openedConnectionsCount: self.openedConnectionsCount, pending: self.pending)
         }
 
-        mutating func testsOnly_setInternalState(_ snapshot: Snapshot) {
+        mutating func testsOnly_setInternalState(_ snapshot: Snapshot<ConnectionType>) {
             self.state = snapshot.state
             self.availableConnections = snapshot.availableConnections
             self.leasedConnections = snapshot.leasedConnections
@@ -109,15 +109,15 @@ extension HTTP1ConnectionProvider {
             return self.openedConnectionsCount == 0 && self.pending == 0
         }
 
-        mutating func acquire(waiter: Waiter) -> Action {
+        mutating func acquire(waiter: Waiter<ConnectionType>) -> Action<ConnectionType> {
             switch self.state {
             case .active:
                 self.pending -= 1
 
                 let (eventLoop, required) = self.resolvePreference(waiter.preference)
                 if required {
                     // If there is an opened connection on the same EL - use it
-                    if let found = self.availableConnections.firstIndex(where: { $0.channel.eventLoop === eventLoop }) {
+                    if let found = self.availableConnections.firstIndex(where: { $0.eventLoop === eventLoop }) {
                         let connection = self.availableConnections.remove(at: found)
                         self.leasedConnections.insert(ConnectionKey(connection))
                         return .lease(connection, waiter)
@@ -151,7 +151,7 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func release(connection: Connection, closing: Bool) -> Action {
+        mutating func release(connection: ConnectionType, closing: Bool) -> Action<ConnectionType> {
             switch self.state {
             case .active:
                 assert(self.leasedConnections.contains(ConnectionKey(connection)))
@@ -161,12 +161,12 @@ extension HTTP1ConnectionProvider {
                         let (eventLoop, required) = self.resolvePreference(waiter.preference)
 
                         // If returned connection is on same EL or we do not require special EL - lease it
-                        if connection.channel.eventLoop === eventLoop || !required {
+                        if connection.eventLoop === eventLoop || !required {
                             return .lease(connection, waiter)
                         }
 
                         // If there is an opened connection on the same loop, lease it and park returned
-                        if let found = self.availableConnections.firstIndex(where: { $0.channel.eventLoop === eventLoop }) {
+                        if let found = self.availableConnections.firstIndex(where: { $0.eventLoop === eventLoop }) {
                             self.leasedConnections.remove(ConnectionKey(connection))
                             let replacement = self.availableConnections.swap(at: found, with: connection)
                             self.leasedConnections.insert(ConnectionKey(replacement))
@@ -203,7 +203,7 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func offer(connection: Connection) -> Action {
+        mutating func offer(connection: ConnectionType) -> Action<ConnectionType> {
             switch self.state {
             case .active:
                 self.leasedConnections.insert(ConnectionKey(connection))
@@ -214,7 +214,7 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func drop(connection: Connection) {
+        mutating func drop(connection: ConnectionType) {
             switch self.state {
             case .active:
                 self.leasedConnections.remove(ConnectionKey(connection))
@@ -223,7 +223,7 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func connectFailed() -> Action {
+        mutating func connectFailed() -> Action<ConnectionType> {
             switch self.state {
             case .active:
                 self.openedConnectionsCount -= 1
@@ -239,7 +239,7 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func remoteClosed(connection: Connection) -> Action {
+        mutating func remoteClosed(connection: ConnectionType) -> Action<ConnectionType> {
             switch self.state {
             case .active:
                 // Connection can be closed remotely while we wait for `.lease` action to complete.
@@ -260,7 +260,7 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func timeout(connection: Connection) -> Action {
+        mutating func timeout(connection: ConnectionType) -> Action<ConnectionType> {
             switch self.state {
             case .active:
                 // We can get timeout and inUse = true when we decided to lease the connection, but this action is not executed yet.
@@ -285,12 +285,12 @@ extension HTTP1ConnectionProvider {
             }
         }
 
-        mutating func processNextWaiter() -> Action {
+        mutating func processNextWaiter() -> Action<ConnectionType> {
             if let waiter = self.waiters.popFirst() {
                 let (eventLoop, required) = self.resolvePreference(waiter.preference)
 
                 // If specific EL is required, we have only two options - find open one or create a new one
-                if required, let found = self.availableConnections.firstIndex(where: { $0.channel.eventLoop === eventLoop }) {
+                if required, let found = self.availableConnections.firstIndex(where: { $0.eventLoop === eventLoop }) {
                     let connection = self.availableConnections.remove(at: found)
                     self.leasedConnections.insert(ConnectionKey(connection))
                     return .lease(connection, waiter)
@@ -313,7 +313,7 @@ extension HTTP1ConnectionProvider {
             return .none
         }
 
-        mutating func close() -> (CircularBuffer<Waiter>, CircularBuffer<Connection>, Set<ConnectionKey>, Bool)? {
+        mutating func close() -> (CircularBuffer<Waiter<ConnectionType>>, CircularBuffer<ConnectionType>, Set<ConnectionKey<ConnectionType>>, Bool)? {
             switch self.state {
             case .active:
                 let waiters = self.waiters

Tests/AsyncHTTPClientTests/ConnectionPoolTests.swift

@@ -28,7 +28,7 @@ class ConnectionPoolTests: XCTestCase {
     var http1ConnectionProvider: HTTP1ConnectionProvider!
 
     func testPending() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
 
         var snapshot = state.testsOnly_getInternalState()
         XCTAssertEqual(0, snapshot.availableConnections.count)
@@ -50,7 +50,7 @@ class ConnectionPoolTests: XCTestCase {
     // MARK: - Acquire Tests
 
     func testAcquireWhenEmpty() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
 
         var snapshot = state.testsOnly_getInternalState()
         XCTAssertEqual(0, snapshot.availableConnections.count)
@@ -155,7 +155,7 @@ class ConnectionPoolTests: XCTestCase {
     // MARK: - Acquire on Specific EL Tests
 
     func testAcquireWhenEmptySpecificEL() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
         var snapshot = state.testsOnly_getInternalState()
 
         XCTAssertEqual(0, snapshot.availableConnections.count)
@@ -301,7 +301,7 @@ class ConnectionPoolTests: XCTestCase {
     // MARK: - Acquire Errors Tests
 
     func testAcquireWhenClosed() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
         _ = state.close()
 
         XCTAssertFalse(state.enqueue())
@@ -317,7 +317,7 @@ class ConnectionPoolTests: XCTestCase {
     }
 
     func testConnectFailedWhenClosed() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
         _ = state.close()
 
         let action = state.connectFailed()
@@ -1222,7 +1222,7 @@ class ConnectionPoolTests: XCTestCase {
     // MARK: - Shutdown tests
 
     func testShutdownOnPendingAndSuccess() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
 
         XCTAssertTrue(state.enqueue())
 
@@ -1261,7 +1261,7 @@ class ConnectionPoolTests: XCTestCase {
     }
 
     func testShutdownOnPendingAndError() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
 
         XCTAssertTrue(state.enqueue())
 
@@ -1298,7 +1298,7 @@ class ConnectionPoolTests: XCTestCase {
     }
 
     func testShutdownTimeout() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
 
         var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
 
@@ -1333,7 +1333,7 @@ class ConnectionPoolTests: XCTestCase {
     }
 
     func testShutdownRemoteClosed() {
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: self.eventLoop)
 
         var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
 

Tests/AsyncHTTPClientTests/ConnectionPoolTestsSupport.swift

@@ -17,10 +17,10 @@ import NIO
 import XCTest
 
 extension ConnectionPoolTests {
-    func buildState(count: Int, release: Bool = true, eventLoop: EventLoop? = nil) -> (HTTP1ConnectionProvider.ConnectionsState, [Connection]) {
+    func buildState(count: Int, release: Bool = true, eventLoop: EventLoop? = nil) -> (HTTP1ConnectionProvider.ConnectionsState<Connection>, [Connection]) {
         let eventLoop = eventLoop ?? self.eventLoop!
 
-        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: eventLoop)
+        var state = HTTP1ConnectionProvider.ConnectionsState<Connection>(eventLoop: eventLoop)
         var items: [Connection] = []
 
         if count == 0 {
@@ -60,7 +60,7 @@ extension ConnectionPoolTests {
     }
 }
 
-func XCTAssertState(_ state: HTTP1ConnectionProvider.ConnectionsState, available: Int, leased: Int, waiters: Int, pending: Int, opened: Int) {
+func XCTAssertState<ConnectionType>(_ state: HTTP1ConnectionProvider.ConnectionsState<ConnectionType>, available: Int, leased: Int, waiters: Int, pending: Int, opened: Int) {
     let snapshot = state.testsOnly_getInternalState()
     XCTAssertEqual(available, snapshot.availableConnections.count)
     XCTAssertEqual(leased, snapshot.leasedConnections.count)
@@ -69,7 +69,7 @@ func XCTAssertState(_ state: HTTP1ConnectionProvider.ConnectionsState, available
     XCTAssertEqual(opened, snapshot.openedConnectionsCount)
 }
 
-func XCTAssertState(_ state: HTTP1ConnectionProvider.ConnectionsState, available: Int, leased: Int, waiters: Int, pending: Int, opened: Int, isLeased connection: Connection) {
+func XCTAssertState<ConnectionType>(_ state: HTTP1ConnectionProvider.ConnectionsState<ConnectionType>, available: Int, leased: Int, waiters: Int, pending: Int, opened: Int, isLeased connection: ConnectionType) {
     let snapshot = state.testsOnly_getInternalState()
     XCTAssertEqual(available, snapshot.availableConnections.count)
     XCTAssertEqual(leased, snapshot.leasedConnections.count)
@@ -79,7 +79,7 @@ func XCTAssertState(_ state: HTTP1ConnectionProvider.ConnectionsState, available
     XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
 }
 
-func XCTAssertState(_ state: HTTP1ConnectionProvider.ConnectionsState, available: Int, leased: Int, waiters: Int, pending: Int, opened: Int, isNotLeased connection: Connection) {
+func XCTAssertState<ConnectionType>(_ state: HTTP1ConnectionProvider.ConnectionsState<ConnectionType>, available: Int, leased: Int, waiters: Int, pending: Int, opened: Int, isNotLeased connection: ConnectionType) {
     let snapshot = state.testsOnly_getInternalState()
     XCTAssertEqual(available, snapshot.availableConnections.count)
     XCTAssertEqual(leased, snapshot.leasedConnections.count)
@@ -100,7 +100,7 @@ func XCTUnwrap<T>(_ value: T?) throws -> T {
 
 struct TempError: Error {}
 
-func XCTAssertStateClose(_ state: HTTP1ConnectionProvider.ConnectionsState, available: Int, leased: Int, waiters: Int, clean: Bool) throws {
+func XCTAssertStateClose<ConnectionType>(_ state: HTTP1ConnectionProvider.ConnectionsState<ConnectionType>, available: Int, leased: Int, waiters: Int, clean: Bool) throws {
     var state = state
 
     let (foundWaiters, foundAvailable, foundLeased, foundClean) = try XCTUnwrap(state.close())