ConnectionPoolTests.swift

//===----------------------------------------------------------------------===//
//
// This source file is part of the AsyncHTTPClient open source project
//
// Copyright (c) 2018-2019 Apple Inc. and the AsyncHTTPClient project authors
// Licensed under Apache License v2.0
//
// See LICENSE.txt for license information
// See CONTRIBUTORS.txt for the list of AsyncHTTPClient project authors
//
// SPDX-License-Identifier: Apache-2.0
//
//===----------------------------------------------------------------------===//

@testable import AsyncHTTPClient
@testable import NIO
import NIOConcurrencyHelpers
import NIOFoundationCompat
import NIOHTTP1
import NIOHTTPCompression
import NIOSSL
import NIOTestUtils
import NIOTransportServices
import XCTest

class ConnectionPoolTests: XCTestCase {
    var eventLoop: EmbeddedEventLoop!
    var http1ConnectionProvider: HTTP1ConnectionProvider!

    struct TempError: Error {}

    func testPending() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)

        var snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)

        XCTAssertTrue(state.enqueue())

        snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(1, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)
    }

    // MARK: - Acquire Tests

    func testAcquireWhenEmpty() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)

        var snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)

        XCTAssertTrue(state.enqueue())
        let action = state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .indifferent))
        switch action {
        case .create(let waiter):
            waiter.promise.fail(TempError())
        default:
            XCTFail("Unexpected action: \(action)")
        }

        snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)
    }

    func testAcquireWhenAvailable() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        XCTAssertTrue(self.http1ConnectionProvider.enqueue())

        let action = self.http1ConnectionProvider.state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .indifferent))
        switch action {
        case .lease(let connection, let waiter):
            waiter.promise.succeed(connection)

            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testAcquireWhenUnavailable() throws {
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.openedConnectionsCount = 8
        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(8, snapshot.openedConnectionsCount)

        XCTAssertTrue(self.http1ConnectionProvider.enqueue())

        let action = self.http1ConnectionProvider.state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .indifferent))
        switch action {
        case .none:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(1, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(8, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        snapshot.openedConnectionsCount = 0
        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)
    }

    // MARK: - Acquire on Specific EL Tests

    func testAcquireWhenEmptySpecificEL() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
        var snapshot = state.testsOnly_getInternalState()

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)

        XCTAssertTrue(state.enqueue())

        let action = state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: self.eventLoop)))
        switch action {
        case .create(let waiter):
            waiter.promise.fail(TempError())
        default:
            XCTFail("Unexpected action: \(action)")
        }

        snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)
    }

    func testAcquireWhenAvailableSpecificEL() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        XCTAssertTrue(self.http1ConnectionProvider.enqueue())

        let action = self.http1ConnectionProvider.state.acquire(waiter: .init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: channel.eventLoop)))
        switch action {
        case .lease(let connection, let waiter):
            waiter.promise.succeed(connection)

            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testAcquireReplace() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 8

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(8, snapshot.openedConnectionsCount)

        XCTAssertTrue(self.http1ConnectionProvider.enqueue())

        let action = self.http1ConnectionProvider.state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: self.eventLoop)))
        switch action {
        case .replace(_, let waiter):
            waiter.promise.fail(TempError())

            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(8, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        snapshot.openedConnectionsCount = 0
        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)
    }

    func testAcquireWhenUnavailableSpecificEL() throws {
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.openedConnectionsCount = 8

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(8, snapshot.openedConnectionsCount)

        XCTAssertTrue(self.http1ConnectionProvider.enqueue())

        let action = self.http1ConnectionProvider.state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: self.eventLoop)))
        switch action {
        case .none:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(1, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(8, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        snapshot.openedConnectionsCount = 0
        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)
    }

    // MARK: - Acquire Errors Tests

    func testAcquireWhenClosed() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
        _ = state.close()

        XCTAssertFalse(state.enqueue())

        let promise = self.eventLoop.makePromise(of: Connection.self)
        let action = state.acquire(waiter: .init(promise: promise, setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .indifferent))
        switch action {
        case .fail(let waiter, let error):
            waiter.promise.fail(error)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testConnectFailedWhenClosed() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
        _ = state.close()

        let action = state.connectFailed()
        switch action {
        case .none:
            break
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    // MARK: - Release Tests

    func testReleaseAliveConnectionEmptyQueue() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .park:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        connection.remoteClosed(logger: HTTPClient.loggingDisabled)
    }

    func testReleaseAliveButClosingConnectionEmptyQueue() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: true)
        switch action {
        case .closeProvider:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(0, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.execute(action, logger: HTTPClient.loggingDisabled)
    }

    func testReleaseInactiveConnectionEmptyQueue() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: true)
        switch action {
        case .closeProvider:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(0, snapshot.openedConnectionsCount)

        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.execute(action, logger: HTTPClient.loggingDisabled)
    }

    func testReleaseInactiveConnectionEmptyQueueHasConnections() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 2
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(2, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: true)
        switch action {
        case .none:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        connection.remoteClosed(logger: HTTPClient.loggingDisabled)
    }

    func testReleaseAliveConnectionHasWaiter() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .indifferent))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .lease(let connection, let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.succeed(connection)
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseInactiveConnectionHasWaitersNoConnections() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .indifferent))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: true)
        switch action {
        case .create(let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            // simulate create -> use -> release cycle
            self.http1ConnectionProvider.connect(.failure(TempError()), waiter: waiter, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseInactiveConnectionHasWaitersHasConnections() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 2
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .indifferent))

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(2, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .lease(let connection, let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.succeed(connection)
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    // MARK: - Release on Specific EL Tests

    func testReleaseAliveConnectionSameELHasWaiterSpecificEL() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: channel.eventLoop)))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .lease(let connection, let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.succeed(connection)
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseAliveConnectionDifferentELNoSameELConnectionsHasWaiterSpecificEL() throws {
        let differentEL = EmbeddedEventLoop()
        defer {
            XCTAssertNoThrow(try differentEL.syncShutdownGracefully())
        }
        let channel = ActiveChannel(eventLoop: differentEL) // Channel on different EL, that's important for the test.
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(of: Connection.self),
                                      setupComplete: self.eventLoop.makeSucceededFuture(()),
                                      preference: .delegateAndChannel(on: self.eventLoop)))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .parkAnd(let connection, .create(let waiter)):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertFalse(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(2, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            // simulate create -> use -> release cycle
            self.http1ConnectionProvider.connect(.failure(TempError()), waiter: waiter, logger: HTTPClient.loggingDisabled)
            connection.remoteClosed(logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseAliveConnectionDifferentELHasSameELConnectionsHasWaiterSpecificEL() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)
        let otherChannel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 2
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: otherChannel.eventLoop)))

        let available = Connection(channel: otherChannel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(2, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .parkAnd(let connection, .lease(let replacement, let waiter)):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertFalse(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(replacement)))
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(2, snapshot.openedConnectionsCount)

            // cleanup
            waiter.promise.succeed(replacement)
            connection.remoteClosed(logger: HTTPClient.loggingDisabled)
            self.http1ConnectionProvider.release(connection: replacement, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseAliveConnectionDifferentELNoSameELConnectionsOnLimitHasWaiterSpecificEL() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)
        let otherChannel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 8
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: otherChannel.eventLoop)))

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(8, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .replace(let connection, let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(8, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.fail(TempError())
            snapshot.openedConnectionsCount = 2
            self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

            snapshot.availableConnections.forEach { $0.remoteClosed(logger: HTTPClient.loggingDisabled) }
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseInactiveConnectionHasWaitersHasSameELConnectionsSpecificEL() throws {
        let channel = EmbeddedChannel()
        let otherChannel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 2
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: otherChannel.eventLoop)))

        let available = Connection(channel: otherChannel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(2, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .lease(let connection, let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(connection === available)
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.succeed(connection)
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testReleaseInactiveConnectionHasWaitersNoSameELConnectionsSpecificEL() throws {
        let channel = EmbeddedChannel()
        let otherChannel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 2
        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.leasedConnections.insert(ConnectionKey(connection))
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: otherChannel.eventLoop)))

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(2, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.release(connection: connection, closing: false)
        switch action {
        case .create(let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(2, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)s
            self.http1ConnectionProvider.connect(.failure(TempError()), waiter: waiter, logger: HTTPClient.loggingDisabled)
            snapshot.availableConnections.forEach { $0.remoteClosed(logger: HTTPClient.loggingDisabled) }
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    // MARK: - Next Waiter Tests

    func testNextWaiterEmptyQueue() throws {
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.processNextWaiter()
        switch action {
        case .closeProvider:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(0, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.execute(action, logger: HTTPClient.loggingDisabled)
    }

    func testNextWaiterEmptyQueueHasConnections() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.processNextWaiter()
        switch action {
        case .none:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            available.remoteClosed(logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testNextWaiterHasWaitersHasConnections() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .indifferent))

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.processNextWaiter()
        switch action {
        case .lease(let connection, let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.fail(TempError())
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testNextWaiterHasWaitersHasSameELConnectionsSpecificEL() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: channel.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: channel.eventLoop)))

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.processNextWaiter()
        switch action {
        case .lease(let connection, let waiter):
            let snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.fail(TempError())
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testNextWaiterHasWaitersHasDifferentELConnectionsSpecificEL() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.waiters.append(.init(promise: channel.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .delegateAndChannel(on: self.eventLoop)))

        let available = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(available)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(1, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.processNextWaiter()
        switch action {
        case .create(let waiter):
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(1, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(2, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            // simulate create -> use -> release cycle
            self.http1ConnectionProvider.connect(.failure(TempError()), waiter: waiter, logger: HTTPClient.loggingDisabled)
            available.remoteClosed(logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    // MARK: - Timeout and Remote Close Tests

    func testTimeoutLeasedConnection() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.timeout(connection: connection)
        switch action {
        case .none:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
    }

    func testTimeoutAvailableConnection() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.availableConnections.append(connection)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.timeout(connection: connection)
        switch action {
        case .closeAnd(_, let after):
            switch after {
            case .closeProvider:
                break
            default:
                XCTFail("Unexpected action: \(action)")
            }
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(0, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.execute(action, logger: HTTPClient.loggingDisabled)
    }

    func testRemoteClosedLeasedConnection() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.remoteClosed(connection: connection)
        switch action {
        case .none:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
    }

    func testRemoteClosedAvailableConnection() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.pending = 0
        snapshot.openedConnectionsCount = 1
        snapshot.availableConnections.append(connection)

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        let action = self.http1ConnectionProvider.state.remoteClosed(connection: connection)
        switch action {
        case .closeProvider:
            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(0, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(0, snapshot.openedConnectionsCount)
        default:
            XCTFail("Unexpected action: \(action)")
        }

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        self.http1ConnectionProvider.execute(action, logger: HTTPClient.loggingDisabled)
    }

    // MARK: - Connection Tests

    func testConnectionReleaseActive() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.openedConnectionsCount = 1
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        connection.release(closing: false, logger: HTTPClient.loggingDisabled)

        // XCTAssertFalse(connection.isInUse)
        snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        // cleanup
        // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
        // (https://github.com/swift-server/async-http-client/issues/234)
        connection.remoteClosed(logger: HTTPClient.loggingDisabled)
    }

    func testConnectionReleaseInactive() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.openedConnectionsCount = 1
        snapshot.leasedConnections.insert(ConnectionKey(connection))

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(1, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        connection.release(closing: true, logger: HTTPClient.loggingDisabled)

        snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)
    }

    func testConnectionRemoteCloseRelease() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        connection.remoteClosed(logger: HTTPClient.loggingDisabled)

        snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)
    }

    func testConnectionTimeoutRelease() throws {
        let channel = EmbeddedChannel()

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        connection.timeout(logger: HTTPClient.loggingDisabled)

        snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
        XCTAssertEqual(0, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(0, snapshot.openedConnectionsCount)
    }

    func testAcquireAvailableBecomesUnavailable() throws {
        let channel = ActiveChannel(eventLoop: self.eventLoop)
        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: channel, provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        self.http1ConnectionProvider.state.testsOnly_setInternalState(snapshot)

        XCTAssertEqual(1, snapshot.availableConnections.count)
        XCTAssertEqual(0, snapshot.leasedConnections.count)
        XCTAssertEqual(0, snapshot.waiters.count)
        XCTAssertEqual(0, snapshot.pending)
        XCTAssertEqual(1, snapshot.openedConnectionsCount)

        XCTAssertTrue(self.http1ConnectionProvider.enqueue())

        let action = self.http1ConnectionProvider.state.acquire(waiter: .init(promise: self.eventLoop.makePromise(), setupComplete: self.eventLoop.makeSucceededFuture(()), preference: .indifferent))
        switch action {
        case .lease(let connection, let waiter):
            // Since this connection is already in use, this should be a no-op and state should not have changed from normal lease
            connection.timeout(logger: HTTPClient.loggingDisabled)

            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(connection.isActiveEstimation)
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // This is unrecoverable, but in this case we create a new connection, so state again should not change, even though release will be called
            // This is important to preventself.http1ConnectionProvider deletion since connection is released and there could be 0 waiters
            connection.remoteClosed(logger: HTTPClient.loggingDisabled)

            snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()
            XCTAssertTrue(snapshot.leasedConnections.contains(ConnectionKey(connection)))
            XCTAssertEqual(0, snapshot.availableConnections.count)
            XCTAssertEqual(1, snapshot.leasedConnections.count)
            XCTAssertEqual(0, snapshot.waiters.count)
            XCTAssertEqual(0, snapshot.pending)
            XCTAssertEqual(1, snapshot.openedConnectionsCount)

            // cleanup
            // this cleanup code needs to go and use HTTP1ConnectionProvider's API instead
            // (https://github.com/swift-server/async-http-client/issues/234)
            waiter.promise.fail(TempError())
            self.http1ConnectionProvider.release(connection: connection, closing: true, logger: HTTPClient.loggingDisabled)
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    // MARK: - Shutdown tests

    func testShutdownOnPendingAndSuccess() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)

        XCTAssertTrue(state.enqueue())

        let connectionPromise = self.eventLoop.makePromise(of: Connection.self)
        let setupPromise = self.eventLoop.makePromise(of: Void.self)
        let waiter = HTTP1ConnectionProvider.Waiter(promise: connectionPromise, setupComplete: setupPromise.futureResult, preference: .indifferent)
        var action = state.acquire(waiter: waiter)

        switch action {
        case .create:
            // expected
            break
        default:
            XCTFail("Unexpected action: \(action)")
        }

        let snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(snapshot.openedConnectionsCount, 1)

        if let (waiters, available, leased, clean) = state.close() {
            XCTAssertTrue(waiters.isEmpty)
            XCTAssertTrue(available.isEmpty)
            XCTAssertTrue(leased.isEmpty)
            XCTAssertFalse(clean)
        } else {
            XCTFail("Expecting snapshot")
        }

        let connection = Connection(channel: EmbeddedChannel(), provider: self.http1ConnectionProvider)

        action = state.offer(connection: connection)
        switch action {
        case .closeAnd(_, let next):
            switch next {
            case .closeProvider:
                // expected
                break
            default:
                XCTFail("Unexpected action: \(action)")
            }
        default:
            XCTFail("Unexpected action: \(action)")
        }

        connectionPromise.fail(TempError())
        setupPromise.succeed(())
    }

    func testShutdownOnPendingAndError() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)

        XCTAssertTrue(state.enqueue())

        let connectionPromise = self.eventLoop.makePromise(of: Connection.self)
        let setupPromise = self.eventLoop.makePromise(of: Void.self)
        let waiter = HTTP1ConnectionProvider.Waiter(promise: connectionPromise, setupComplete: setupPromise.futureResult, preference: .indifferent)
        var action = state.acquire(waiter: waiter)

        switch action {
        case .create:
            // expected
            break
        default:
            XCTFail("Unexpected action: \(action)")
        }

        let snapshot = state.testsOnly_getInternalState()
        XCTAssertEqual(snapshot.openedConnectionsCount, 1)

        if let (waiters, available, leased, clean) = state.close() {
            XCTAssertTrue(waiters.isEmpty)
            XCTAssertTrue(available.isEmpty)
            XCTAssertTrue(leased.isEmpty)
            XCTAssertFalse(clean)
        } else {
            XCTFail("Expecting snapshot")
        }

        action = state.connectFailed()
        switch action {
        case .closeProvider:
            // expected
            break
        default:
            XCTFail("Unexpected action: \(action)")
        }

        connectionPromise.fail(TempError())
        setupPromise.succeed(())
    }

    func testShutdownTimeout() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: EmbeddedChannel(), provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        state.testsOnly_setInternalState(snapshot)

        if let (waiters, available, leased, clean) = state.close() {
            XCTAssertTrue(waiters.isEmpty)
            XCTAssertFalse(available.isEmpty)
            XCTAssertTrue(leased.isEmpty)
            XCTAssertTrue(clean)
        } else {
            XCTFail("Expecting snapshot")
        }

        let action = state.timeout(connection: connection)
        switch action {
        case .closeAnd(_, let next):
            switch next {
            case .closeProvider:
                // expected
                break
            default:
                XCTFail("Unexpected action: \(action)")
            }
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    func testShutdownRemoteClosed() {
        var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)

        var snapshot = self.http1ConnectionProvider.state.testsOnly_getInternalState()

        let connection = Connection(channel: EmbeddedChannel(), provider: self.http1ConnectionProvider)
        snapshot.availableConnections.append(connection)
        snapshot.openedConnectionsCount = 1

        state.testsOnly_setInternalState(snapshot)

        if let (waiters, available, leased, clean) = state.close() {
            XCTAssertTrue(waiters.isEmpty)
            XCTAssertFalse(available.isEmpty)
            XCTAssertTrue(leased.isEmpty)
            XCTAssertTrue(clean)
        } else {
            XCTFail("Expecting snapshot")
        }

        let action = state.remoteClosed(connection: connection)
        switch action {
        case .closeProvider:
            // expected
            break
        default:
            XCTFail("Unexpected action: \(action)")
        }
    }

    // MARK: - Helpers

    override func setUp() {
        XCTAssertNil(self.eventLoop)
        XCTAssertNil(self.http1ConnectionProvider)
        self.eventLoop = EmbeddedEventLoop()
        XCTAssertNoThrow(self.http1ConnectionProvider = try HTTP1ConnectionProvider(key: .init(.init(url: "http://some.test")),
                                                                                    eventLoop: self.eventLoop,
                                                                                    configuration: .init(),
                                                                                    pool: .init(configuration: .init(),
                                                                                                backgroundActivityLogger: HTTPClient.loggingDisabled),
                                                                                    backgroundActivityLogger: HTTPClient.loggingDisabled))
    }

    override func tearDown() {
        XCTAssertNotNil(self.eventLoop)
        XCTAssertNotNil(self.http1ConnectionProvider)
        XCTAssertNoThrow(try self.http1ConnectionProvider.close().wait())
        XCTAssertNoThrow(try self.eventLoop.syncShutdownGracefully())
        self.eventLoop = nil
        self.http1ConnectionProvider = nil
    }
}

class ActiveChannel: Channel, ChannelCore {
    struct NotImplementedError: Error {}

    func localAddress0() throws -> SocketAddress {
        throw NotImplementedError()
    }

    func remoteAddress0() throws -> SocketAddress {
        throw NotImplementedError()
    }

    func register0(promise: EventLoopPromise<Void>?) {
        promise?.fail(NotImplementedError())
    }

    func bind0(to: SocketAddress, promise: EventLoopPromise<Void>?) {
        promise?.fail(NotImplementedError())
    }

    func connect0(to: SocketAddress, promise: EventLoopPromise<Void>?) {
        promise?.fail(NotImplementedError())
    }

    func write0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        promise?.fail(NotImplementedError())
    }

    func flush0() {}

    func read0() {}

    func close0(error: Error, mode: CloseMode, promise: EventLoopPromise<Void>?) {
        promise?.succeed(())
    }

    func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
        promise?.fail(NotImplementedError())
    }

    func channelRead0(_: NIOAny) {}

    func errorCaught0(error: Error) {}

    var allocator: ByteBufferAllocator
    var closeFuture: EventLoopFuture<Void>
    var eventLoop: EventLoop

    var localAddress: SocketAddress?
    var remoteAddress: SocketAddress?
    var parent: Channel?
    var isWritable: Bool = true
    var isActive: Bool = true

    init(eventLoop: EmbeddedEventLoop) {
        self.allocator = ByteBufferAllocator()
        self.eventLoop = eventLoop
        self.closeFuture = self.eventLoop.makeSucceededFuture(())
    }

    var _channelCore: ChannelCore {
        return self
    }

    var pipeline: ChannelPipeline {
        return ChannelPipeline(channel: self)
    }

    func setOption<Option>(_ option: Option, value: Option.Value) -> EventLoopFuture<Void> where Option: ChannelOption {
        preconditionFailure("Not implemented")
    }

    func getOption<Option>(_: Option) -> EventLoopFuture<Option.Value> where Option: ChannelOption {
        preconditionFailure("Not implemented")
    }
}