Merge branch 'master' into assume_chunked_on_zero_length_stream

Author: weissi

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.16.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.18.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.7.0"),
         .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.3.0"),
         .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.5.1"),

Sources/AsyncHTTPClient/ConnectionsState.swift

@@ -229,7 +229,10 @@ extension HTTP1ConnectionProvider {
                 self.openedConnectionsCount -= 1
                 return self.processNextWaiter()
             case .closed:
-                assertionFailure("should not happen")
+                // This can happen in the following scenario: user initiates a connection that will fail to connect,
+                // user calls `syncShutdown` before we received an error from the bootstrap. In this scenario,
+                // pool will be `.closed` but connection will be still in the process of being established/failed,
+                // so then this process finishes, it will get to this point.
                 return .none
             }
         }

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -925,6 +925,7 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
         case uncleanShutdown
         case traceRequestWithBody
         case invalidHeaderFieldNames([String])
+        case bodyLengthMismatch
     }
 
     private var code: Code
@@ -969,10 +970,12 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     public static let redirectLimitReached = HTTPClientError(code: .redirectLimitReached)
     /// Redirect Cycle detected.
     public static let redirectCycleDetected = HTTPClientError(code: .redirectCycleDetected)
-    /// Unclean shutdown
+    /// Unclean shutdown.
     public static let uncleanShutdown = HTTPClientError(code: .uncleanShutdown)
-    /// A body was sent in a request with method TRACE
+    /// A body was sent in a request with method TRACE.
     public static let traceRequestWithBody = HTTPClientError(code: .traceRequestWithBody)
-    /// Header field names contain invalid characters
+    /// Header field names contain invalid characters.
     public static func invalidHeaderFieldNames(_ names: [String]) -> HTTPClientError { return HTTPClientError(code: .invalidHeaderFieldNames(names)) }
+    /// Body length is not equal to `Content-Length`.
+    public static let bodyLengthMismatch = HTTPClientError(code: .bodyLengthMismatch)
 }

Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -77,9 +77,7 @@ extension HTTPClient {
         ///     - data: Body `Data` representation.
         public static func data(_ data: Data) -> Body {
             return Body(length: data.count) { writer in
-                var buffer = ByteBufferAllocator().buffer(capacity: data.count)
-                buffer.writeBytes(data)
-                return writer.write(.byteBuffer(buffer))
+                writer.write(.byteBuffer(ByteBuffer(bytes: data)))
             }
         }
 
@@ -89,9 +87,7 @@ extension HTTPClient {
         ///     - string: Body `String` representation.
         public static func string(_ string: String) -> Body {
             return Body(length: string.utf8.count) { writer in
-                var buffer = ByteBufferAllocator().buffer(capacity: string.utf8.count)
-                buffer.writeString(string)
-                return writer.write(.byteBuffer(buffer))
+                writer.write(.byteBuffer(ByteBuffer(string: string)))
             }
         }
     }
@@ -641,7 +637,7 @@ internal class TaskHandler<Delegate: HTTPClientResponseDelegate>: RemovableChann
         case head
         case redirected(HTTPResponseHead, URL)
         case body
-        case end
+        case endOrError
     }
 
     let task: HTTPClient.Task<Delegate.Response>
@@ -651,6 +647,8 @@ internal class TaskHandler<Delegate: HTTPClientResponseDelegate>: RemovableChann
     let logger: Logger // We are okay to store the logger here because a TaskHandler is just for one request.
 
     var state: State = .idle
+    var expectedBodyLength: Int?
+    var actualBodyLength: Int = 0
     var pendingRead = false
     var mayRead = true
     var closing = false {
@@ -771,16 +769,21 @@ extension TaskHandler: ChannelDuplexHandler {
                                    uri: request.uri)
         var headers = request.headers
 
-        if !request.headers.contains(name: "Host") {
-            headers.add(name: "Host", value: request.host)
+        if !request.headers.contains(name: "host") {
+            let port = request.port
+            var host = request.host
+            if !(port == 80 && request.scheme == "http"), !(port == 443 && request.scheme == "https") {
+                host += ":\(port)"
+            }
+            headers.add(name: "host", value: host)
         }
 
         do {
             try headers.validate(method: request.method, body: request.body)
         } catch {
             promise?.fail(error)
             self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
-            self.state = .end
+            self.state = .endOrError
             return
         }
 
@@ -794,12 +797,23 @@ extension TaskHandler: ChannelDuplexHandler {
         assert(head.version == HTTPVersion(major: 1, minor: 1),
                "Sending a request in HTTP version \(head.version) which is unsupported by the above `if`")
 
+        let contentLengths = head.headers[canonicalForm: "content-length"]
+        assert(contentLengths.count <= 1)
+
+        self.expectedBodyLength = contentLengths.first.flatMap { Int($0) }
+
         context.write(wrapOutboundOut(.head(head))).map {
             self.callOutToDelegateFireAndForget(value: head, self.delegate.didSendRequestHead)
         }.flatMap {
             self.writeBody(request: request, context: context)
         }.flatMap {
             context.eventLoop.assertInEventLoop()
+            if let expectedBodyLength = self.expectedBodyLength, expectedBodyLength != self.actualBodyLength {
+                self.state = .endOrError
+                let error = HTTPClientError.bodyLengthMismatch
+                self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
+                return context.eventLoop.makeFailedFuture(error)
+            }
             return context.writeAndFlush(self.wrapOutboundOut(.end(nil)))
         }.map {
             context.eventLoop.assertInEventLoop()
@@ -808,10 +822,10 @@ extension TaskHandler: ChannelDuplexHandler {
         }.flatMapErrorThrowing { error in
             context.eventLoop.assertInEventLoop()
             switch self.state {
-            case .end:
+            case .endOrError:
                 break
             default:
-                self.state = .end
+                self.state = .endOrError
                 self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
             }
             throw error
@@ -828,9 +842,11 @@ extension TaskHandler: ChannelDuplexHandler {
                 let promise = self.task.eventLoop.makePromise(of: Void.self)
                 // All writes have to be switched to the channel EL if channel and task ELs differ
                 if context.eventLoop.inEventLoop {
+                    self.actualBodyLength += part.readableBytes
                     context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
                 } else {
                     context.eventLoop.execute {
+                        self.actualBodyLength += part.readableBytes
                         context.writeAndFlush(self.wrapOutboundOut(.body(part)), promise: promise)
                     }
                 }
@@ -893,12 +909,12 @@ extension TaskHandler: ChannelDuplexHandler {
         case .end:
             switch self.state {
             case .redirected(let head, let redirectURL):
-                self.state = .end
+                self.state = .endOrError
                 self.task.releaseAssociatedConnection(delegateType: Delegate.self, closing: self.closing).whenSuccess {
                     self.redirectHandler?.redirect(status: head.status, to: redirectURL, promise: self.task.promise)
                 }
             default:
-                self.state = .end
+                self.state = .endOrError
                 self.callOutToDelegate(promise: self.task.promise, self.delegate.didFinishRequest)
             }
         }
@@ -913,14 +929,14 @@ extension TaskHandler: ChannelDuplexHandler {
                 context.read()
             }
         case .failure(let error):
-            self.state = .end
+            self.state = .endOrError
             self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
         }
     }
 
     func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
         if (event as? IdleStateHandler.IdleStateEvent) == .read {
-            self.state = .end
+            self.state = .endOrError
             let error = HTTPClientError.readTimeout
             self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
         } else {
@@ -930,7 +946,7 @@ extension TaskHandler: ChannelDuplexHandler {
 
     func triggerUserOutboundEvent(context: ChannelHandlerContext, event: Any, promise: EventLoopPromise<Void>?) {
         if (event as? TaskCancelEvent) != nil {
-            self.state = .end
+            self.state = .endOrError
             let error = HTTPClientError.cancelled
             self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
             promise?.succeed(())
@@ -941,10 +957,10 @@ extension TaskHandler: ChannelDuplexHandler {
 
     func channelInactive(context: ChannelHandlerContext) {
         switch self.state {
-        case .end:
+        case .endOrError:
             break
         case .body, .head, .idle, .redirected, .sent:
-            self.state = .end
+            self.state = .endOrError
             let error = HTTPClientError.remoteConnectionClosed
             self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
         }
@@ -955,7 +971,7 @@ extension TaskHandler: ChannelDuplexHandler {
         switch error {
         case NIOSSLError.uncleanShutdown:
             switch self.state {
-            case .end:
+            case .endOrError:
                 /// Some HTTP Servers can 'forget' to respond with CloseNotify when client is closing connection,
                 /// this could lead to incomplete SSL shutdown. But since request is already processed, we can ignore this error.
                 break
@@ -964,11 +980,11 @@ extension TaskHandler: ChannelDuplexHandler {
                 /// We can also ignore this error like `.end`.
                 break
             default:
-                self.state = .end
+                self.state = .endOrError
                 self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
             }
         default:
-            self.state = .end
+            self.state = .endOrError
             self.failTaskAndNotifyDelegate(error: error, self.delegate.didReceiveError)
         }
     }

Tests/AsyncHTTPClientTests/ConnectionPoolTests+XCTest.swift

@@ -34,6 +34,7 @@ extension ConnectionPoolTests {
             ("testAcquireReplace", testAcquireReplace),
             ("testAcquireWhenUnavailableSpecificEL", testAcquireWhenUnavailableSpecificEL),
             ("testAcquireWhenClosed", testAcquireWhenClosed),
+            ("testConnectFailedWhenClosed", testConnectFailedWhenClosed),
             ("testReleaseAliveConnectionEmptyQueue", testReleaseAliveConnectionEmptyQueue),
             ("testReleaseAliveButClosingConnectionEmptyQueue", testReleaseAliveButClosingConnectionEmptyQueue),
             ("testReleaseInactiveConnectionEmptyQueue", testReleaseInactiveConnectionEmptyQueue),

Tests/AsyncHTTPClientTests/ConnectionPoolTests.swift

@@ -304,9 +304,7 @@ class ConnectionPoolTests: XCTestCase {
 
     func testAcquireWhenClosed() {
         var state = HTTP1ConnectionProvider.ConnectionsState(eventLoop: self.eventLoop)
-        var snapshot = state.testsOnly_getInternalState()
-        snapshot.state = .closed
-        state.testsOnly_setInternalState(snapshot)
+        _ = state.close()
 
         XCTAssertFalse(state.enqueue())
 
@@ -320,6 +318,19 @@ class ConnectionPoolTests: XCTestCase {
         }
     }
 
+    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 {

Tests/AsyncHTTPClientTests/HTTPClientInternalTests+XCTest.swift

@@ -27,6 +27,7 @@ extension HTTPClientInternalTests {
         return [
             ("testHTTPPartsHandler", testHTTPPartsHandler),
             ("testBadHTTPRequest", testBadHTTPRequest),
+            ("testHostPort", testHostPort),
             ("testHTTPPartsHandlerMultiBody", testHTTPPartsHandlerMultiBody),
             ("testProxyStreaming", testProxyStreaming),
             ("testProxyStreamingFailure", testProxyStreamingFailure),