More tests

Author: fabianfett

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTP1.1/HTTP1Connection.swift

@@ -87,7 +87,13 @@ final class HTTP1Connection {
     }
 
     func execute(request: HTTPExecutableRequest) {
-        self.channel.write(request, promise: nil)
+        if self.channel.eventLoop.inEventLoop {
+            self.execute0(request: request)
+        } else {
+            self.channel.eventLoop.execute {
+                self.execute0(request: request)
+            }
+        }
     }
 
     func cancel() {
@@ -101,4 +107,12 @@ final class HTTP1Connection {
     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)
+    }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests+XCTest.swift

@@ -25,6 +25,10 @@ import XCTest
 extension HTTP1ClientChannelHandlerTests {
     static var allTests: [(String, (HTTP1ClientChannelHandlerTests) -> () throws -> Void)] {
         return [
+            ("testResponseBackpressure", testResponseBackpressure),
+            ("testWriteBackpressure", testWriteBackpressure),
+            ("testClientHandlerCancelsRequestIfWeWantToShutdown", testClientHandlerCancelsRequestIfWeWantToShutdown),
+            ("testIdleReadTimeout", testIdleReadTimeout),
         ]
     }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests.swift

@@ -104,6 +104,186 @@ class HTTP1ClientChannelHandlerTests: XCTestCase {
         XCTAssertNoThrow(try requestBag.task.futureResult.wait())
     }
 
+    func testWriteBackpressure() {
+        class TestWriter {
+            let eventLoop: EventLoop
+
+            let parts: Int
+
+            var finishFuture: EventLoopFuture<Void> { self.finishPromise.futureResult }
+            private let finishPromise: EventLoopPromise<Void>
+            private(set) var written: Int = 0
+
+            private var channelIsWritable: Bool = false
+
+            init(eventLoop: EventLoop, parts: Int) {
+                self.eventLoop = eventLoop
+                self.parts = parts
+
+                self.finishPromise = eventLoop.makePromise(of: Void.self)
+            }
+
+            func start(writer: HTTPClient.Body.StreamWriter) -> EventLoopFuture<Void> {
+                func recursive() {
+                    XCTAssert(self.eventLoop.inEventLoop)
+                    XCTAssert(self.channelIsWritable)
+                    if self.written == self.parts {
+                        self.finishPromise.succeed(())
+                    } else {
+                        self.eventLoop.execute {
+                            let future = writer.write(.byteBuffer(.init(bytes: [0, 1])))
+                            self.written += 1
+                            future.whenComplete { result in
+                                switch result {
+                                case .success:
+                                    recursive()
+                                case .failure(let error):
+                                    XCTFail("Unexpected error: \(error)")
+                                }
+                            }
+                        }
+                    }
+                }
+
+                recursive()
+
+                return self.finishFuture
+            }
+
+            func writabilityChanged(_ newValue: Bool) {
+                self.channelIsWritable = newValue
+            }
+        }
+
+        let embedded = EmbeddedChannel()
+        let testWriter = TestWriter(eventLoop: embedded.eventLoop, parts: 50)
+        var maybeTestUtils: HTTP1TestTools?
+        XCTAssertNoThrow(maybeTestUtils = try embedded.setupHTTP1Connection())
+        guard let testUtils = maybeTestUtils else { return XCTFail("Expected connection setup works") }
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "http://localhost/", method: .POST, body: .stream(length: 100) { writer in
+            testWriter.start(writer: writer)
+        }))
+        guard let request = maybeRequest else { return XCTFail("Expected to be able to create a request") }
+
+        let delegate = ResponseAccumulator(request: request)
+        var maybeRequestBag: RequestBag<ResponseAccumulator>?
+        XCTAssertNoThrow(maybeRequestBag = try RequestBag(
+            request: request,
+            eventLoopPreference: .delegate(on: embedded.eventLoop),
+            task: .init(eventLoop: embedded.eventLoop, logger: testUtils.logger),
+            redirectHandler: nil,
+            connectionDeadline: .now() + .seconds(30),
+            idleReadTimeout: .milliseconds(200),
+            delegate: delegate
+        ))
+        guard let requestBag = maybeRequestBag else { return XCTFail("Expected to be able to create a request bag") }
+
+        // the handler only writes once the channel is writable
+        embedded.isWritable = false
+        testWriter.writabilityChanged(false)
+        embedded.pipeline.fireChannelWritabilityChanged()
+        testUtils.connection.execute(request: requestBag)
+
+        XCTAssertEqual(try embedded.readOutbound(as: HTTPClientRequestPart.self), .none)
+
+        embedded.isWritable = true
+        testWriter.writabilityChanged(true)
+        embedded.pipeline.fireChannelWritabilityChanged()
+
+        XCTAssertNoThrow(try embedded.receiveHeadAndVerify {
+            XCTAssertEqual($0.method, .POST)
+            XCTAssertEqual($0.uri, "/")
+            XCTAssertEqual($0.headers.first(name: "host"), "localhost")
+            XCTAssertEqual($0.headers.first(name: "content-length"), "100")
+        })
+
+        // the next body write will be executed once we tick the el. before we make the channel
+        // unwritable
+
+        for index in 0..<50 {
+            embedded.isWritable = false
+            testWriter.writabilityChanged(false)
+            embedded.pipeline.fireChannelWritabilityChanged()
+
+            XCTAssertEqual(testWriter.written, index)
+
+            embedded.embeddedEventLoop.run()
+
+            XCTAssertNoThrow(try embedded.receiveBodyAndVerify {
+                XCTAssertEqual($0.readableBytes, 2)
+            })
+
+            XCTAssertEqual(testWriter.written, index + 1)
+
+            embedded.isWritable = true
+            testWriter.writabilityChanged(true)
+            embedded.pipeline.fireChannelWritabilityChanged()
+        }
+
+        embedded.embeddedEventLoop.run()
+        XCTAssertNoThrow(try embedded.receiveEnd())
+
+        let responseHead = HTTPResponseHead(version: .http1_1, status: .ok)
+        XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.head(responseHead)))
+        embedded.read()
+
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionClosed, 0)
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionReleased, 0)
+        XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.end(nil)))
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionClosed, 0)
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionReleased, 1)
+
+        XCTAssertNoThrow(try requestBag.task.futureResult.wait())
+    }
+
+    func testClientHandlerCancelsRequestIfWeWantToShutdown() {
+        let embedded = EmbeddedChannel()
+        var maybeTestUtils: HTTP1TestTools?
+        XCTAssertNoThrow(maybeTestUtils = try embedded.setupHTTP1Connection())
+        guard let testUtils = maybeTestUtils else { return XCTFail("Expected connection setup works") }
+
+        var maybeRequest: HTTPClient.Request?
+        XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "http://localhost/"))
+        guard let request = maybeRequest else { return XCTFail("Expected to be able to create a request") }
+
+        let delegate = ResponseAccumulator(request: request)
+        var maybeRequestBag: RequestBag<ResponseAccumulator>?
+        XCTAssertNoThrow(maybeRequestBag = try RequestBag(
+            request: request,
+            eventLoopPreference: .delegate(on: embedded.eventLoop),
+            task: .init(eventLoop: embedded.eventLoop, logger: testUtils.logger),
+            redirectHandler: nil,
+            connectionDeadline: .now() + .seconds(30),
+            idleReadTimeout: .milliseconds(200),
+            delegate: delegate
+        ))
+        guard let requestBag = maybeRequestBag else { return XCTFail("Expected to be able to create a request bag") }
+
+        testUtils.connection.execute(request: requestBag)
+
+        XCTAssertNoThrow(try embedded.receiveHeadAndVerify {
+            XCTAssertEqual($0.method, .GET)
+            XCTAssertEqual($0.uri, "/")
+            XCTAssertEqual($0.headers.first(name: "host"), "localhost")
+        })
+        XCTAssertNoThrow(try embedded.receiveEnd())
+
+        XCTAssertTrue(embedded.isActive)
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionClosed, 0)
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionReleased, 0)
+        testUtils.connection.cancel()
+        XCTAssertFalse(embedded.isActive)
+        embedded.embeddedEventLoop.run()
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionClosed, 1)
+        XCTAssertEqual(testUtils.connectionDelegate.hitConnectionReleased, 0)
+
+        XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
+            XCTAssertEqual($0 as? HTTPClientError, .cancelled)
+        }
+    }
+
     func testIdleReadTimeout() {
         let embedded = EmbeddedChannel()
         var maybeTestUtils: HTTP1TestTools?

Diff for: Tests/AsyncHTTPClientTests/HTTP1ConnectionTests+XCTest.swift

@@ -29,6 +29,7 @@ extension HTTP1ConnectionTests {
             ("testCreateNewConnectionWithoutDecompression", testCreateNewConnectionWithoutDecompression),
             ("testCreateNewConnectionFailureClosedIO", testCreateNewConnectionFailureClosedIO),
             ("testGETRequest", testGETRequest),
+            ("testConnectionClosesOnCloseHeader", testConnectionClosesOnCloseHeader),
         ]
     }
 }

Diff for: Tests/AsyncHTTPClientTests/HTTP1ConnectionTests.swift

@@ -185,6 +185,72 @@ class HTTP1ConnectionTests: XCTestCase {
         // connection is closed
         XCTAssertNoThrow(try XCTUnwrap(delegate.closePromise).futureResult.wait())
     }
+
+    func testConnectionClosesOnCloseHeader() {
+        let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        let eventLoop = eventLoopGroup.next()
+        defer { XCTAssertNoThrow(try eventLoopGroup.syncShutdownGracefully()) }
+
+        let closeOnRequest = (30...100).randomElement()!
+        let httpBin = HTTPBin(handlerFactory: { _ in SuddenlySendsCloseHeaderChannel(closeOnRequest: closeOnRequest) })
+
+        var maybeChannel: Channel?
+
+        XCTAssertNoThrow(maybeChannel = try ClientBootstrap(group: eventLoop).connect(host: "localhost", port: httpBin.port).wait())
+        let connectionDelegate = MockConnectionDelegate()
+        let logger = Logger(label: "test")
+        var maybeConnection: HTTP1Connection?
+        XCTAssertNoThrow(maybeConnection = try eventLoop.submit { try HTTP1Connection(
+            channel: XCTUnwrap(maybeChannel),
+            connectionID: 0,
+            configuration: .init(),
+            delegate: connectionDelegate,
+            logger: logger
+        ) }.wait())
+        guard let connection = maybeConnection else { return XCTFail("Expected to have a connection here") }
+
+        var counter = 0
+        while true {
+            counter += 1
+
+            var maybeRequest: HTTPClient.Request?
+            XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "http://localhost/"))
+            guard let request = maybeRequest else { return XCTFail("Expected to be able to create a request") }
+
+            let delegate = ResponseAccumulator(request: request)
+            var maybeRequestBag: RequestBag<ResponseAccumulator>?
+            XCTAssertNoThrow(maybeRequestBag = try RequestBag(
+                request: request,
+                eventLoopPreference: .delegate(on: eventLoopGroup.next()),
+                task: .init(eventLoop: eventLoopGroup.next(), logger: logger),
+                redirectHandler: nil,
+                connectionDeadline: .now() + .seconds(30),
+                idleReadTimeout: nil,
+                delegate: delegate
+            ))
+            guard let requestBag = maybeRequestBag else { return XCTFail("Expected to be able to create a request bag") }
+
+            connection.execute(request: requestBag)
+
+            var response: HTTPClient.Response?
+            if counter <= closeOnRequest {
+                XCTAssertNoThrow(response = try requestBag.task.futureResult.wait())
+                XCTAssertEqual(response?.status, .ok)
+
+                if response?.headers.first(name: "connection") == "close" {
+                    XCTAssertEqual(closeOnRequest, counter)
+                    XCTAssertEqual(maybeChannel?.isActive, false)
+                }
+            } else {
+                // io on close channel leads to error
+                XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
+                    XCTAssertEqual($0 as? ChannelError, .ioOnClosedChannel)
+                }
+
+                break // the loop
+            }
+        }
+    }
 }
 
 class MockHTTP1ConnectionDelegate: HTTP1ConnectionDelegate {
@@ -199,3 +265,38 @@ class MockHTTP1ConnectionDelegate: HTTP1ConnectionDelegate {
         self.closePromise?.succeed(())
     }
 }
+
+class SuddenlySendsCloseHeaderChannel: ChannelInboundHandler {
+    typealias InboundIn = HTTPServerRequestPart
+    typealias OutboundOut = HTTPServerResponsePart
+
+    var counter = 1
+    let closeOnRequest: Int
+
+    init(closeOnRequest: Int) {
+        self.closeOnRequest = closeOnRequest
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        switch self.unwrapInboundIn(data) {
+        case .head(let head):
+            XCTAssertLessThanOrEqual(self.counter, self.closeOnRequest)
+            XCTAssertTrue(head.headers.contains(name: "host"))
+            XCTAssertEqual(head.method, .GET)
+        case .body:
+            break
+        case .end:
+            if self.closeOnRequest == self.counter {
+                context.write(self.wrapOutboundOut(.head(.init(version: .http1_1, status: .ok, headers: ["connection": "close"]))), promise: nil)
+                context.write(self.wrapOutboundOut(.end(nil)), promise: nil)
+                context.flush()
+                self.counter += 1
+            } else {
+                context.write(self.wrapOutboundOut(.head(.init(version: .http1_1, status: .ok))), promise: nil)
+                context.write(self.wrapOutboundOut(.end(nil)), promise: nil)
+                context.flush()
+                self.counter += 1
+            }
+        }
+    }
+}

Diff for: Tests/AsyncHTTPClientTests/RequestBagTests+XCTest.swift

@@ -30,6 +30,7 @@ extension RequestBagTests {
             ("testCancelFailsTaskBeforeRequestIsSent", testCancelFailsTaskBeforeRequestIsSent),
             ("testCancelFailsTaskAfterRequestIsSent", testCancelFailsTaskAfterRequestIsSent),
             ("testCancelFailsTaskWhenTaskIsQueued", testCancelFailsTaskWhenTaskIsQueued),
+            ("testFailsTaskWhenTaskIsWaitingForMoreFromServer", testFailsTaskWhenTaskIsWaitingForMoreFromServer),
             ("testHTTPUploadIsCancelledEvenThoughRequestSucceeds", testHTTPUploadIsCancelledEvenThoughRequestSucceeds),
         ]
     }