Support informational response heads

Package.swift

@@ -21,7 +21,7 @@ let package = Package(
         .library(name: "AsyncHTTPClient", targets: ["AsyncHTTPClient"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.32.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.34.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.14.1"),
         .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.18.2"),
         .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.10.0"),

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

@@ -116,7 +116,16 @@ final class HTTP1Connection {
 
         do {
             let sync = self.channel.pipeline.syncOperations
-            try sync.addHTTPClientHandlers()
+
+            // We can not use `sync.addHTTPClientHandlers()`, as we want to explicitly set the
+            // `.informationalResponseStrategy` for the decoder.
+            let requestEncoder = HTTPRequestEncoder()
+            let responseDecoder = HTTPResponseDecoder(
+                leftOverBytesStrategy: .dropBytes,
+                informationalResponseStrategy: .forward
+            )
+            try sync.addHandler(requestEncoder)
+            try sync.addHandler(ByteToMessageHandler(responseDecoder))
 
             if case .enabled(let limit) = configuration.decompression {
                 let decompressHandler = NIOHTTPResponseDecompressor(limit: limit)

Sources/AsyncHTTPClient/ConnectionPool/HTTPRequestStateMachine.swift

@@ -529,13 +529,7 @@ struct HTTPRequestStateMachine {
             preconditionFailure("How can we receive a response head before sending a request head ourselves. Invalid state: \(self.state)")
 
         case .running(_, .waitingForHead):
-            // If we receive a http response header with a status code of 1xx, we ignore the header
-            // except for 101, which we consume.
-            // If the remote closes the connection after sending a 1xx (not 101) response head, we
-            // will receive a response end from the parser. We need to protect against this case.
-            let error = HTTPClientError.httpEndReceivedAfterHeadWith1xx
-            self.state = .failed(error)
-            return .failRequest(error, .close)
+            preconditionFailure("How can we receive a response end, if we haven't a received a head. Invalid state: \(self.state)")
 
         case .running(.streaming(let expectedBodyLength, let sentBodyBytes, let producerState), .receivingBody(let head, var responseStreamState))
             where head.status.code < 300:

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -996,5 +996,6 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     ///  - Tasks are not processed fast enough on the existing connections, to process all waiters in time
     public static let getConnectionFromPoolTimeout = HTTPClientError(code: .getConnectionFromPoolTimeout)
 
+    @available(*, deprecated, message: "AsyncHTTPClient now correctly supports informational headers. For this reason `httpEndReceivedAfterHeadWith1xx` will not be thrown anymore.")
     public static let httpEndReceivedAfterHeadWith1xx = HTTPClientError(code: .httpEndReceivedAfterHeadWith1xx)
 }

Tests/AsyncHTTPClientTests/HTTP1ConnectionStateMachineTests.swift

@@ -265,7 +265,7 @@ class HTTP1ConnectionStateMachineTests: XCTestCase {
         XCTAssertEqual(newRequestAction, .sendRequestHead(requestHead, startBody: false))
         let responseHead = HTTPResponseHead(version: .http1_1, status: .init(statusCode: 103, reasonPhrase: "Early Hints"))
         XCTAssertEqual(state.channelRead(.head(responseHead)), .wait)
-        XCTAssertEqual(state.channelRead(.end(nil)), .failRequest(HTTPClientError.httpEndReceivedAfterHeadWith1xx, .close))
+        XCTAssertEqual(state.channelInactive(), .failRequest(HTTPClientError.remoteConnectionClosed, .none))
     }
 }
 

Tests/AsyncHTTPClientTests/HTTP1ConnectionTests+XCTest.swift

@@ -33,7 +33,7 @@ extension HTTP1ConnectionTests {
             ("testConnectionClosesOnRandomlyAppearingCloseHeader", testConnectionClosesOnRandomlyAppearingCloseHeader),
             ("testConnectionClosesAfterTheRequestWithoutHavingSentAnCloseHeader", testConnectionClosesAfterTheRequestWithoutHavingSentAnCloseHeader),
             ("testConnectionIsClosedAfterSwitchingProtocols", testConnectionIsClosedAfterSwitchingProtocols),
-            ("testConnectionDoesntCrashAfterConnectionCloseAndEarlyHints", testConnectionDoesntCrashAfterConnectionCloseAndEarlyHints),
+            ("testConnectionDropAfterEarlyHints", testConnectionDropAfterEarlyHints),
             ("testConnectionIsClosedIfResponseIsReceivedBeforeRequest", testConnectionIsClosedIfResponseIsReceivedBeforeRequest),
             ("testDoubleHTTPResponseLine", testDoubleHTTPResponseLine),
             ("testDownloadStreamingBackpressure", testDownloadStreamingBackpressure),

Tests/AsyncHTTPClientTests/HTTP1ConnectionTests.swift

@@ -430,7 +430,7 @@ class HTTP1ConnectionTests: XCTestCase {
         XCTAssertEqual(response?.body, nil)
     }
 
-    func testConnectionDoesntCrashAfterConnectionCloseAndEarlyHints() {
+    func testConnectionDropAfterEarlyHints() {
         let embedded = EmbeddedChannel()
         let logger = Logger(label: "test.http1.connection")
 
@@ -481,25 +481,15 @@ class HTTP1ConnectionTests: XCTestCase {
         XCTAssertEqual(connectionDelegate.hitConnectionReleased, 0)
         XCTAssertNoThrow(try embedded.writeInbound(ByteBuffer(string: responseString)))
 
-        if !embedded.isActive {
-            // behavior before https://github.com/apple/swift-nio/pull/1984
-            embedded.embeddedEventLoop.run() // tick once to run futures.
-            XCTAssertEqual(connectionDelegate.hitConnectionClosed, 1)
-            XCTAssertEqual(connectionDelegate.hitConnectionReleased, 0)
+        XCTAssertTrue(embedded.isActive, "The connection remains active after the informational response head")
+        XCTAssertNoThrow(try embedded.close().wait(), "the connection was closed")
 
-            XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
-                XCTAssertEqual($0 as? HTTPClientError, .httpEndReceivedAfterHeadWith1xx)
-            }
-        } else {
-            // behavior after https://github.com/apple/swift-nio/pull/1984
-            XCTAssertNoThrow(try embedded.close().wait())
-            embedded.embeddedEventLoop.run() // tick once to run futures.
-            XCTAssertEqual(connectionDelegate.hitConnectionClosed, 1)
-            XCTAssertEqual(connectionDelegate.hitConnectionReleased, 0)
-
-            XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
-                XCTAssertEqual($0 as? HTTPClientError, .remoteConnectionClosed)
-            }
+        embedded.embeddedEventLoop.run() // tick once to run futures.
+        XCTAssertEqual(connectionDelegate.hitConnectionClosed, 1)
+        XCTAssertEqual(connectionDelegate.hitConnectionReleased, 0)
+
+        XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
+            XCTAssertEqual($0 as? HTTPClientError, .remoteConnectionClosed)
         }
     }
 

Tests/AsyncHTTPClientTests/HTTPClientInformationalResponsesTests+XCTest.swift

@@ -0,0 +1,32 @@
+//===----------------------------------------------------------------------===//
+//
+// 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
+//
+//===----------------------------------------------------------------------===//
+//
+// HTTPClientInformationalResponsesTests+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 HTTPClientReproTests {
+    static var allTests: [(String, (HTTPClientReproTests) -> () throws -> Void)] {
+        return [
+            ("testServerSends100ContinueFirst", testServerSends100ContinueFirst),
+            ("testServerSendsSwitchingProtocols", testServerSendsSwitchingProtocols),
+        ]
+    }
+}

Tests/AsyncHTTPClientTests/HTTPClientInformationalResponsesTests.swift

@@ -0,0 +1,122 @@
+//===----------------------------------------------------------------------===//
+//
+// 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 AsyncHTTPClient
+import Logging
+import NIOCore
+import NIOHTTP1
+import XCTest
+
+final class HTTPClientReproTests: XCTestCase {
+    func testServerSends100ContinueFirst() {
+        final class HTTPInformationalResponseHandler: ChannelInboundHandler {
+            typealias InboundIn = HTTPServerRequestPart
+            typealias OutboundOut = HTTPServerResponsePart
+
+            func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+                switch self.unwrapInboundIn(data) {
+                case .head:
+                    context.writeAndFlush(self.wrapOutboundOut(.head(.init(version: .http1_1, status: .continue))), promise: nil)
+                case .body:
+                    break
+                case .end:
+                    context.write(self.wrapOutboundOut(.head(.init(version: .http1_1, status: .ok))), promise: nil)
+                    context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
+                }
+            }
+        }
+
+        let client = HTTPClient(eventLoopGroupProvider: .createNew)
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+
+        let httpBin = HTTPBin(.http1_1(ssl: false, compress: false)) { _ in
+            HTTPInformationalResponseHandler()
+        }
+
+        let body = #"{"foo": "bar"}"#
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(
+            url: "http://localhost:\(httpBin.port)/",
+            method: .POST,
+            headers: [
+                "Content-Type": "application/json",
+            ],
+            body: .string(body)
+        ))
+        guard let request = maybeRequest else { return XCTFail("Expected to have a request here") }
+
+        var logger = Logger(label: "test")
+        logger.logLevel = .trace
+
+        var response: HTTPClient.Response?
+        XCTAssertNoThrow(response = try client.execute(request: request, logger: logger).wait())
+        XCTAssertEqual(response?.status, .ok)
+    }
+
+    func testServerSendsSwitchingProtocols() {
+        final class HTTPInformationalResponseHandler: ChannelInboundHandler {
+            typealias InboundIn = HTTPServerRequestPart
+            typealias OutboundOut = HTTPServerResponsePart
+
+            func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+                switch self.unwrapInboundIn(data) {
+                case .head:
+                    let head = HTTPResponseHead(version: .http1_1, status: .switchingProtocols, headers: [
+                        "Connection": "Upgrade",
+                        "Upgrade": "Websocket",
+                    ])
+                    let body = context.channel.allocator.buffer(string: "foo bar")
+
+                    context.write(self.wrapOutboundOut(.head(head)), promise: nil)
+                    context.write(self.wrapOutboundOut(.body(.byteBuffer(body))), promise: nil)
+                    // we purposefully don't send an `.end` here.
+                    context.flush()
+                case .body:
+                    break
+                case .end:
+                    break
+                }
+            }
+        }
+
+        let client = HTTPClient(eventLoopGroupProvider: .createNew)
+        defer { XCTAssertNoThrow(try client.syncShutdown()) }
+
+        let httpBin = HTTPBin(.http1_1(ssl: false, compress: false)) { _ in
+            HTTPInformationalResponseHandler()
+        }
+
+        let body = #"{"foo": "bar"}"#
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(
+            url: "http://localhost:\(httpBin.port)/",
+            method: .POST,
+            headers: [
+                "Content-Type": "application/json",
+            ],
+            body: .string(body)
+        ))
+        guard let request = maybeRequest else { return XCTFail("Expected to have a request here") }
+
+        var logger = Logger(label: "test")
+        logger.logLevel = .trace
+
+        var response: HTTPClient.Response?
+        XCTAssertNoThrow(response = try client.execute(request: request, logger: logger).wait())
+        XCTAssertEqual(response?.status, .switchingProtocols)
+        XCTAssertNil(response?.body)
+    }
+}

Tests/LinuxMain.swift

@@ -37,6 +37,7 @@ import XCTest
         testCase(HTTPClientCookieTests.allTests),
         testCase(HTTPClientInternalTests.allTests),
         testCase(HTTPClientNIOTSTests.allTests),
+        testCase(HTTPClientReproTests.allTests),
         testCase(HTTPClientSOCKSTests.allTests),
         testCase(HTTPClientTests.allTests),
         testCase(HTTPConnectionPoolTests.allTests),