Add HTTP1Connection

Sources/AsyncHTTPClient/ConnectionPool/HTTP1.1/HTTP1Connection.swift

@@ -0,0 +1,118 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the AsyncHTTPClient open source project
+//
+// Copyright (c) 2021 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
+//
+//===----------------------------------------------------------------------===//
+
+import Logging
+import NIO
+import NIOHTTP1
+import NIOHTTPCompression
+
+protocol HTTP1ConnectionDelegate {
+    func http1ConnectionReleased(_: HTTP1Connection)
+    func http1ConnectionClosed(_: HTTP1Connection)
+}
+
+final class HTTP1Connection {
+    let channel: Channel
+
+    /// the connection's delegate, that will be informed about connection close and connection release
+    /// (ready to run next request).
+    let delegate: HTTP1ConnectionDelegate
+
+    enum State {
+        case active
+        case closed
+    }
+
+    private var state: State = .active
+
+    let id: HTTPConnectionPool.Connection.ID
+
+    init(channel: Channel,
+         connectionID: HTTPConnectionPool.Connection.ID,
+         configuration: HTTPClient.Configuration,
+         delegate: HTTP1ConnectionDelegate,
+         logger: Logger) throws {
+        channel.eventLoop.assertInEventLoop()
+
+        // let's add the channel handlers needed for h1
+        self.channel = channel
+        self.id = connectionID
+        self.delegate = delegate
+
+        // all properties are set here. Therefore the connection is fully initialized. If we
+        // run into an error, here we need to do the state handling ourselfes.
+
+        do {
+            let sync = channel.pipeline.syncOperations
+            try sync.addHTTPClientHandlers()
+
+            if case .enabled(let limit) = configuration.decompression {
+                let decompressHandler = NIOHTTPResponseDecompressor(limit: limit)
+                try sync.addHandler(decompressHandler)
+            }
+
+            let channelHandler = HTTP1ClientChannelHandler(
+                connection: self,
+                eventLoop: channel.eventLoop,
+                logger: logger
+            )
+            try sync.addHandler(channelHandler)
+
+            // with this we create an intended retain cycle...
+            self.channel.closeFuture.whenComplete { _ in
+                self.state = .closed
+                self.delegate.http1ConnectionClosed(self)
+            }
+        } catch {
+            self.state = .closed
+            throw error
+        }
+    }
+
+    deinit {
+        guard case .closed = self.state else {
+            preconditionFailure("Connection must be closed, before we can deinit it")
+        }
+    }
+
+    func execute(request: HTTPExecutableRequest) {
+        if self.channel.eventLoop.inEventLoop {
+            self.execute0(request: request)
+        } else {
+            self.channel.eventLoop.execute {
+                self.execute0(request: request)
+            }
+        }
+    }
+
+    func cancel() {
+        self.channel.triggerUserOutboundEvent(HTTPConnectionEvent.cancelRequest, promise: nil)
+    }
+
+    func close() -> EventLoopFuture<Void> {
+        return self.channel.close()
+    }
+
+    func taskCompleted() {
+        self.delegate.http1ConnectionReleased(self)
+    }
+
+    private func execute0(request: HTTPExecutableRequest) {
+        guard self.channel.isActive else {
+            return request.fail(ChannelError.ioOnClosedChannel)
+        }
+
+        self.channel.write(request, promise: nil)
+    }
+}

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionEvent.swift

@@ -0,0 +1,17 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the AsyncHTTPClient open source project
+//
+// Copyright (c) 2021 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
+//
+//===----------------------------------------------------------------------===//
+
+enum HTTPConnectionEvent {
+    case cancelRequest
+}

Sources/AsyncHTTPClient/ConnectionPool/HTTPExecutableRequest.swift

@@ -120,7 +120,7 @@ import NIOHTTP1
 ///      - The executor may have received an error in thread A that it passes along to the request.
 ///        After having passed on the error, the executor considers the request done and releases
 ///        the request's reference.
-///      - The request may issue a call to `writeRequestBodyPart(_: IOData, task: HTTPExecutingRequest)`
+///      - The request may issue a call to `writeRequestBodyPart(_: IOData, task: HTTPExecutableRequest)`
 ///        on thread B in the same moment the request error above occurred. For this reason it may
 ///        happen that the executor receives, the invocation of `writeRequestBodyPart` after it has
 ///        failed the request.

Sources/AsyncHTTPClient/RequestBag+StateMachine.swift

@@ -497,9 +497,9 @@ extension RequestBag.StateMachine {
         case .executing(let executor, let requestState, .buffering(_, next: .eof)):
             self.state = .executing(executor, requestState, .buffering(.init(), next: .error(error)))
             return .cancelExecutor(executor)
-        case .executing(let executor, let requestState, .buffering(_, next: .askExecutorForMore)):
-            self.state = .executing(executor, requestState, .buffering(.init(), next: .error(error)))
-            return .cancelExecutor(executor)
+        case .executing(let executor, _, .buffering(_, next: .askExecutorForMore)):
+            self.state = .finished(error: error)
+            return .failTask(nil, executor)
         case .executing(let executor, _, .buffering(_, next: .error(_))):
             // this would override another error, let's keep the first one
             return .cancelExecutor(executor)

Tests/AsyncHTTPClientTests/EmbeddedChannel+HTTPConvenience.swift

@@ -0,0 +1,114 @@
+//===----------------------------------------------------------------------===//
+//
+// 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
+import Logging
+import NIO
+import NIOHTTP1
+
+extension EmbeddedChannel {
+    public func receiveHeadAndVerify(_ verify: (HTTPRequestHead) throws -> Void = { _ in }) throws {
+        let part = try self.readOutbound(as: HTTPClientRequestPart.self)
+        switch part {
+        case .head(let head):
+            try verify(head)
+        case .body, .end:
+            throw HTTP1EmbeddedChannelError(reason: "Expected .head but got '\(part!)'")
+        case .none:
+            throw HTTP1EmbeddedChannelError(reason: "Nothing in buffer")
+        }
+    }
+
+    public func receiveBodyAndVerify(_ verify: (IOData) throws -> Void = { _ in }) throws {
+        let part = try self.readOutbound(as: HTTPClientRequestPart.self)
+        switch part {
+        case .body(let iodata):
+            try verify(iodata)
+        case .head, .end:
+            throw HTTP1EmbeddedChannelError(reason: "Expected .head but got '\(part!)'")
+        case .none:
+            throw HTTP1EmbeddedChannelError(reason: "Nothing in buffer")
+        }
+    }
+
+    public func receiveEnd() throws {
+        let part = try self.readOutbound(as: HTTPClientRequestPart.self)
+        switch part {
+        case .end:
+            break
+        case .head, .body:
+            throw HTTP1EmbeddedChannelError(reason: "Expected .head but got '\(part!)'")
+        case .none:
+            throw HTTP1EmbeddedChannelError(reason: "Nothing in buffer")
+        }
+    }
+}
+
+struct HTTP1TestTools {
+    let connection: HTTP1Connection
+    let connectionDelegate: MockConnectionDelegate
+    let readEventHandler: ReadEventHitHandler
+    let logger: Logger
+}
+
+extension EmbeddedChannel {
+    func setupHTTP1Connection() throws -> HTTP1TestTools {
+        let logger = Logger(label: "test")
+        let readEventHandler = ReadEventHitHandler()
+
+        try self.pipeline.syncOperations.addHandler(readEventHandler)
+        try self.connect(to: .makeAddressResolvingHost("localhost", port: 0)).wait()
+
+        let connectionDelegate = MockConnectionDelegate()
+        let connection = try HTTP1Connection(
+            channel: self,
+            connectionID: 1,
+            configuration: .init(),
+            delegate: connectionDelegate,
+            logger: logger
+        )
+
+        // remove HTTP client encoder and decoder
+
+        let decoder = try self.pipeline.syncOperations.handler(type: ByteToMessageHandler<HTTPResponseDecoder>.self)
+        let encoder = try self.pipeline.syncOperations.handler(type: HTTPRequestEncoder.self)
+
+        let removeDecoderFuture = self.pipeline.removeHandler(decoder)
+        let removeEncoderFuture = self.pipeline.removeHandler(encoder)
+
+        self.embeddedEventLoop.run()
+
+        try removeDecoderFuture.wait()
+        try removeEncoderFuture.wait()
+
+        return .init(
+            connection: connection,
+            connectionDelegate: connectionDelegate,
+            readEventHandler: readEventHandler,
+            logger: logger
+        )
+    }
+}
+
+public struct HTTP1EmbeddedChannelError: Error, Hashable, CustomStringConvertible {
+    public var reason: String
+
+    public init(reason: String) {
+        self.reason = reason
+    }
+
+    public var description: String {
+        return self.reason
+    }
+}

Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests+XCTest.swift

@@ -0,0 +1,34 @@
+//===----------------------------------------------------------------------===//
+//
+// 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
+//
+//===----------------------------------------------------------------------===//
+//
+// HTTP1ClientChannelHandlerTests+XCTest.swift
+//
+import XCTest
+
+///
+/// NOTE: This file was generated by generate_linux_tests.rb
+///
+/// Do NOT edit this file directly as it will be regenerated automatically when needed.
+///
+
+extension HTTP1ClientChannelHandlerTests {
+    static var allTests: [(String, (HTTP1ClientChannelHandlerTests) -> () throws -> Void)] {
+        return [
+            ("testResponseBackpressure", testResponseBackpressure),
+            ("testWriteBackpressure", testWriteBackpressure),
+            ("testClientHandlerCancelsRequestIfWeWantToShutdown", testClientHandlerCancelsRequestIfWeWantToShutdown),
+            ("testIdleReadTimeout", testIdleReadTimeout),
+        ]
+    }
+}