From 03fefc4ddca332b154478e7686875d2b651c6dba Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Wed, 10 Nov 2021 10:26:39 +0100 Subject: [PATCH 1/9] http2 integration tests --- .../HTTP2ClientTests+XCTest.swift | 4 + .../HTTP2ClientTests.swift | 178 +++++++++++++++++- .../HTTPClientTestUtils.swift | 17 +- 3 files changed, 195 insertions(+), 4 deletions(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests+XCTest.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests+XCTest.swift index ffe9c14a1..753cf9e37 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests+XCTest.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests+XCTest.swift @@ -29,6 +29,10 @@ extension HTTP2ClientTests { ("testConcurrentRequests", testConcurrentRequests), ("testConcurrentRequestsFromDifferentThreads", testConcurrentRequestsFromDifferentThreads), ("testConcurrentRequestsWorkWithRequiredEventLoop", testConcurrentRequestsWorkWithRequiredEventLoop), + ("testUncleanShutdownCancelsExecutingAndQueuedTasks", testUncleanShutdownCancelsExecutingAndQueuedTasks), + ("testCancelingRunningRequest", testCancelingRunningRequest), + ("testStressCancelingRunningRequestFromDifferentThreads", testStressCancelingRunningRequestFromDifferentThreads), + ("testPlatformConnectErrorIsForwardedOnTimeout", testPlatformConnectErrorIsForwardedOnTimeout), ] } } diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index bef964577..530018655 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -37,6 +37,17 @@ class HTTP2ClientTests: XCTestCase { ) } + func makeClientWithActiveHTTP2Connection( + to bin: HTTPBin + ) -> HTTPClient { + let client = self.makeDefaultHTTPClient() + var response: HTTPClient.Response? + XCTAssertNoThrow(response = try client.get(url: "https://localhost:\(bin.port)/get").wait()) + XCTAssertEqual(.ok, response?.status) + XCTAssertEqual(response?.version, .http2) + return client + } + func testSimpleGet() { let bin = HTTPBin(.http2(compress: false)) defer { XCTAssertNoThrow(try bin.shutdown()) } @@ -67,7 +78,7 @@ class HTTP2ClientTests: XCTestCase { func testConcurrentRequestsFromDifferentThreads() { let bin = HTTPBin(.http2(compress: false)) defer { XCTAssertNoThrow(try bin.shutdown()) } - let client = self.makeDefaultHTTPClient() + let client = self.makeClientWithActiveHTTP2Connection(to: bin) defer { XCTAssertNoThrow(try client.syncShutdown()) } let numberOfWorkers = 20 let numberOfRequestsPerWorkers = 20 @@ -92,7 +103,7 @@ class HTTP2ClientTests: XCTestCase { for _ in 0..] = [] + XCTAssertNoThrow(results = try EventLoopFuture + .whenAllComplete(responses, on: client.eventLoopGroup.next()) + .timeout(after: .seconds(2)) + .wait()) + + for result in results { + switch result { + case .success: + XCTFail("Shouldn't succeed") + case .failure(let error): + if let clientError = error as? HTTPClientError, clientError == .cancelled { + continue + } else { + XCTFail("Unexpected error: \(error)") + } + } + } + } + + func testCancelingRunningRequest() { + let bin = HTTPBin(.http2(compress: false)) + defer { XCTAssertNoThrow(try bin.shutdown()) } + let client = self.makeClientWithActiveHTTP2Connection(to: bin) + defer { XCTAssertNoThrow(try client.syncShutdown()) } + + var maybeRequest: HTTPClient.Request? + XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)/sendheaderandwait")) + guard let request = maybeRequest else { return } + + var task: HTTPClient.Task! + let delegate = TestHTTPDelegate() + delegate.stateDidChangeCallback = { state in + guard case .head = state else { return } + // request is definitely running because we just received a head from the server + task.cancel() + } + task = client.execute( + request: request, + delegate: delegate + ) + + XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait(), "Should fail") { error in + guard case let error = error as? HTTPClientError, error == .cancelled else { + return XCTFail("Should fail with cancelled") + } + } + } + + func testStressCancelingRunningRequestFromDifferentThreads() { + let bin = HTTPBin(.http2(compress: false)) + defer { XCTAssertNoThrow(try bin.shutdown()) } + let client = self.makeClientWithActiveHTTP2Connection(to: bin) + defer { XCTAssertNoThrow(try client.syncShutdown()) } + let cancelPool = MultiThreadedEventLoopGroup(numberOfThreads: 10) + defer { XCTAssertNoThrow(try cancelPool.syncShutdownGracefully()) } + + var maybeRequest: HTTPClient.Request? + XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)/sendheaderandwait")) + guard let request = maybeRequest else { return } + + let tasks = (0..<100).map { _ -> HTTPClient.Task in + var task: HTTPClient.Task! + let delegate = TestHTTPDelegate() + delegate.stateDidChangeCallback = { state in + guard case .head = state else { return } + // request is definitely running because we just received a head from the server + cancelPool.next().execute { + // canceling from a different thread + task.cancel() + } + } + task = client.execute( + request: request, + delegate: delegate + ) + return task + } + + for task in tasks { + XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait(), "Should fail") { error in + guard case let error = error as? HTTPClientError, error == .cancelled else { + return XCTFail("Should fail with cancelled") + } + } + } + } + + func testPlatformConnectErrorIsForwardedOnTimeout() { + let bin = HTTPBin(.http2(compress: false)) + + let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 2) + let el1 = clientGroup.next() + let el2 = clientGroup.next() + defer { XCTAssertNoThrow(try clientGroup.syncShutdownGracefully()) } + var tlsConfig = TLSConfiguration.makeClientConfiguration() + tlsConfig.certificateVerification = .none + let client = HTTPClient( + eventLoopGroupProvider: .shared(clientGroup), + configuration: HTTPClient.Configuration( + tlsConfiguration: tlsConfig, + timeout: .init(connect: .milliseconds(1000)), + httpVersion: .automatic + ), + backgroundActivityLogger: Logger(label: "HTTPClient", factory: StreamLogHandler.standardOutput(label:)) + ) + defer { XCTAssertNoThrow(try client.syncShutdown()) } + + var maybeRequest1: HTTPClient.Request? + XCTAssertNoThrow(maybeRequest1 = try HTTPClient.Request(url: "https://localhost:\(bin.port)/get")) + guard let request1 = maybeRequest1 else { return } + + let task1 = client.execute(request: request1, delegate: ResponseAccumulator(request: request1), eventLoop: .delegateAndChannel(on: el1)) + var response1: ResponseAccumulator.Response? + XCTAssertNoThrow(response1 = try task1.wait()) + + XCTAssertEqual(.ok, response1?.status) + XCTAssertEqual(response1?.version, .http2) + let serverPort = bin.port + XCTAssertNoThrow(try bin.shutdown()) + // client is now in HTTP/2 state and the HTTPBin is closed + // start a new server on the old port which closes all connections immediately + let serverGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) + defer { XCTAssertNoThrow(try serverGroup.syncShutdownGracefully()) } + var maybeServer: Channel? + XCTAssertNoThrow(maybeServer = try ServerBootstrap(group: serverGroup) + .serverChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1) + .childChannelInitializer { channel in + channel.close() + } + .childChannelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1) + .bind(host: "0.0.0.0", port: serverPort) + .wait()) + guard let server = maybeServer else { return } + defer { XCTAssertNoThrow(try server.close().wait()) } + + var maybeRequest2: HTTPClient.Request? + XCTAssertNoThrow(maybeRequest2 = try HTTPClient.Request(url: "https://localhost:\(serverPort)/")) + guard let request2 = maybeRequest2 else { return } + + let task2 = client.execute(request: request2, delegate: ResponseAccumulator(request: request2), eventLoop: .delegateAndChannel(on: el2)) + XCTAssertThrowsError(try task2.wait()) { error in + XCTAssertNil( + error as? HTTPClientError, + "error should be some platform specific error that the connection is closed/reset by the other side" + ) + } + } } diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift index c83bca0b7..051d38801 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift @@ -49,11 +49,16 @@ func getDefaultEventLoopGroup(numberOfThreads: Int) -> EventLoopGroup { class TestHTTPDelegate: HTTPClientResponseDelegate { typealias Response = Void - init(backpressureEventLoop: EventLoop? = nil) { + init( + backpressureEventLoop: EventLoop? = nil, + stateDidChangeCallback: ((State) -> Void)? = nil + ) { self.backpressureEventLoop = backpressureEventLoop + self.stateDidChangeCallback = stateDidChangeCallback } var backpressureEventLoop: EventLoop? + var stateDidChangeCallback: ((State) -> Void)? enum State { case idle @@ -63,7 +68,11 @@ class TestHTTPDelegate: HTTPClientResponseDelegate { case error(Error) } - var state = State.idle + var state = State.idle { + didSet { + self.stateDidChangeCallback?(self.state) + } + } func didReceiveHead(task: HTTPClient.Task, _ head: HTTPResponseHead) -> EventLoopFuture { self.state = .head(head) @@ -804,6 +813,10 @@ internal final class HTTPBinHandler: ChannelInboundHandler { builder.add(buff) self.resps.append(builder) return + case "/sendheaderandwait": + // sends some headers and waits indefinitely afterwards + context.writeAndFlush(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .ok))), promise: nil) + return case "/wait": return case "/close": From 88fdd59bf9b83dbf71f77b7aec5e571410d0271c Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Wed, 10 Nov 2021 11:53:27 +0100 Subject: [PATCH 2/9] fix tests on linux we needed an event loop group which outlives the http client --- .../AsyncHTTPClientTests/HTTP2ClientTests.swift | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 530018655..725d80b16 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -24,11 +24,13 @@ import NIOSSL import XCTest class HTTP2ClientTests: XCTestCase { - func makeDefaultHTTPClient() -> HTTPClient { + func makeDefaultHTTPClient( + eventLoopGroupProvider: HTTPClient.EventLoopGroupProvider = .createNew + ) -> HTTPClient { var tlsConfig = TLSConfiguration.makeClientConfiguration() tlsConfig.certificateVerification = .none return HTTPClient( - eventLoopGroupProvider: .createNew, + eventLoopGroupProvider: eventLoopGroupProvider, configuration: HTTPClient.Configuration( tlsConfiguration: tlsConfig, httpVersion: .automatic @@ -38,9 +40,10 @@ class HTTP2ClientTests: XCTestCase { } func makeClientWithActiveHTTP2Connection( - to bin: HTTPBin + to bin: HTTPBin, + eventLoopGroupProvider: HTTPClient.EventLoopGroupProvider = .createNew ) -> HTTPClient { - let client = self.makeDefaultHTTPClient() + let client = self.makeDefaultHTTPClient(eventLoopGroupProvider: eventLoopGroupProvider) var response: HTTPClient.Response? XCTAssertNoThrow(response = try client.get(url: "https://localhost:\(bin.port)/get").wait()) XCTAssertEqual(.ok, response?.status) @@ -202,7 +205,9 @@ class HTTP2ClientTests: XCTestCase { func testUncleanShutdownCancelsExecutingAndQueuedTasks() { let bin = HTTPBin(.http2(compress: false)) defer { XCTAssertNoThrow(try bin.shutdown()) } - let client = self.makeClientWithActiveHTTP2Connection(to: bin) + let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) + defer { XCTAssertNoThrow(try clientGroup.syncShutdownGracefully()) } + let client = self.makeClientWithActiveHTTP2Connection(to: bin, eventLoopGroupProvider: .shared(clientGroup)) // start 20 requests which are guaranteed to never get any response // 10 of them will executed and the other 10 will be queued @@ -215,7 +220,7 @@ class HTTP2ClientTests: XCTestCase { var results: [Result] = [] XCTAssertNoThrow(results = try EventLoopFuture - .whenAllComplete(responses, on: client.eventLoopGroup.next()) + .whenAllComplete(responses, on: clientGroup.next()) .timeout(after: .seconds(2)) .wait()) From be120b392a5ced366f156c24bcf636afc491cd29 Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Wed, 10 Nov 2021 17:07:27 +0100 Subject: [PATCH 3/9] Fix review comments --- .../HTTP2ClientTests.swift | 73 ++++++++++++------- .../HTTPClientTestUtils.swift | 4 - 2 files changed, 45 insertions(+), 32 deletions(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 725d80b16..7a81051cc 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -19,6 +19,7 @@ #endif import Logging import NIOCore +import NIOHTTP1 import NIOPosix import NIOSSL import XCTest @@ -229,29 +230,23 @@ class HTTP2ClientTests: XCTestCase { case .success: XCTFail("Shouldn't succeed") case .failure(let error): - if let clientError = error as? HTTPClientError, clientError == .cancelled { - continue - } else { - XCTFail("Unexpected error: \(error)") - } + XCTAssertEqual(error as? HTTPClientError, .cancelled) } } } func testCancelingRunningRequest() { - let bin = HTTPBin(.http2(compress: false)) + let bin = HTTPBin(.http2(compress: false)) { _ in SendHeaderAndWaitChannelHandler() } defer { XCTAssertNoThrow(try bin.shutdown()) } - let client = self.makeClientWithActiveHTTP2Connection(to: bin) + let client = self.makeDefaultHTTPClient() defer { XCTAssertNoThrow(try client.syncShutdown()) } var maybeRequest: HTTPClient.Request? - XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)/sendheaderandwait")) + XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)")) guard let request = maybeRequest else { return } - var task: HTTPClient.Task! - let delegate = TestHTTPDelegate() - delegate.stateDidChangeCallback = { state in - guard case .head = state else { return } + var task: HTTPClient.Task! + let delegate = HeadReceivedCallback { _ in // request is definitely running because we just received a head from the server task.cancel() } @@ -260,30 +255,26 @@ class HTTP2ClientTests: XCTestCase { delegate: delegate ) - XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait(), "Should fail") { error in - guard case let error = error as? HTTPClientError, error == .cancelled else { - return XCTFail("Should fail with cancelled") - } + XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait()) { + XCTAssertEqual($0 as? HTTPClientError, .cancelled) } } func testStressCancelingRunningRequestFromDifferentThreads() { - let bin = HTTPBin(.http2(compress: false)) + let bin = HTTPBin(.http2(compress: false)) { _ in SendHeaderAndWaitChannelHandler() } defer { XCTAssertNoThrow(try bin.shutdown()) } - let client = self.makeClientWithActiveHTTP2Connection(to: bin) + let client = self.makeDefaultHTTPClient() defer { XCTAssertNoThrow(try client.syncShutdown()) } let cancelPool = MultiThreadedEventLoopGroup(numberOfThreads: 10) defer { XCTAssertNoThrow(try cancelPool.syncShutdownGracefully()) } var maybeRequest: HTTPClient.Request? - XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)/sendheaderandwait")) + XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "https://localhost:\(bin.port)")) guard let request = maybeRequest else { return } let tasks = (0..<100).map { _ -> HTTPClient.Task in - var task: HTTPClient.Task! - let delegate = TestHTTPDelegate() - delegate.stateDidChangeCallback = { state in - guard case .head = state else { return } + var task: HTTPClient.Task! + let delegate = HeadReceivedCallback { _ in // request is definitely running because we just received a head from the server cancelPool.next().execute { // canceling from a different thread @@ -298,17 +289,14 @@ class HTTP2ClientTests: XCTestCase { } for task in tasks { - XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait(), "Should fail") { error in - guard case let error = error as? HTTPClientError, error == .cancelled else { - return XCTFail("Should fail with cancelled") - } + XCTAssertThrowsError(try task.futureResult.timeout(after: .seconds(2)).wait()) { + XCTAssertEqual($0 as? HTTPClientError, .cancelled) } } } func testPlatformConnectErrorIsForwardedOnTimeout() { let bin = HTTPBin(.http2(compress: false)) - let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 2) let el1 = clientGroup.next() let el2 = clientGroup.next() @@ -367,3 +355,32 @@ class HTTP2ClientTests: XCTestCase { } } } + +private final class HeadReceivedCallback: HTTPClientResponseDelegate { + typealias Response = Void + private let didReceiveHeadCallback: (HTTPResponseHead) -> Void + init(didReceiveHead: @escaping (HTTPResponseHead) -> Void) { + self.didReceiveHeadCallback = didReceiveHead + } + + func didReceiveHead(task: HTTPClient.Task, _ head: HTTPResponseHead) -> EventLoopFuture { + self.didReceiveHeadCallback(head) + return task.eventLoop.makeSucceededVoidFuture() + } + + func didFinishRequest(task: HTTPClient.Task) throws {} +} + +/// sends some headers and waits indefinitely afterwards +private final class SendHeaderAndWaitChannelHandler: ChannelInboundHandler { + typealias InboundIn = HTTPServerRequestPart + typealias OutboundOut = HTTPServerResponsePart + + func channelRead(context: ChannelHandlerContext, data: NIOAny) { + context.writeAndFlush(wrapOutboundOut(.head(HTTPResponseHead( + version: HTTPVersion(major: 1, minor: 1), + status: .ok + )) + ), promise: nil) + } +} diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift index 051d38801..092da52c7 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift @@ -813,10 +813,6 @@ internal final class HTTPBinHandler: ChannelInboundHandler { builder.add(buff) self.resps.append(builder) return - case "/sendheaderandwait": - // sends some headers and waits indefinitely afterwards - context.writeAndFlush(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .ok))), promise: nil) - return case "/wait": return case "/close": From 235b223ee1938c602eff14802b3daa0f308165c2 Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Thu, 11 Nov 2021 08:39:26 +0100 Subject: [PATCH 4/9] use default client --- Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 7a81051cc..240ba4a9c 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -82,7 +82,7 @@ class HTTP2ClientTests: XCTestCase { func testConcurrentRequestsFromDifferentThreads() { let bin = HTTPBin(.http2(compress: false)) defer { XCTAssertNoThrow(try bin.shutdown()) } - let client = self.makeClientWithActiveHTTP2Connection(to: bin) + let client = self.makeDefaultHTTPClient() defer { XCTAssertNoThrow(try client.syncShutdown()) } let numberOfWorkers = 20 let numberOfRequestsPerWorkers = 20 From e20c28d2173fa5d860fa9de6b83f809eb7e21c94 Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Thu, 11 Nov 2021 10:52:46 +0100 Subject: [PATCH 5/9] add documentation --- Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 240ba4a9c..a9ab7772e 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -208,6 +208,8 @@ class HTTP2ClientTests: XCTestCase { defer { XCTAssertNoThrow(try bin.shutdown()) } let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) defer { XCTAssertNoThrow(try clientGroup.syncShutdownGracefully()) } + // we need an active connection to guarantee that requests are executed immediately + // without waiting for connection establishment let client = self.makeClientWithActiveHTTP2Connection(to: bin, eventLoopGroupProvider: .shared(clientGroup)) // start 20 requests which are guaranteed to never get any response From a73726db5201fb856d7085b038e1f954b1cdfdcd Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Thu, 11 Nov 2021 10:55:06 +0100 Subject: [PATCH 6/9] revert changes to TestHTTPDelegate --- .../AsyncHTTPClientTests/HTTPClientTestUtils.swift | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift index 092da52c7..c83bca0b7 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift @@ -49,16 +49,11 @@ func getDefaultEventLoopGroup(numberOfThreads: Int) -> EventLoopGroup { class TestHTTPDelegate: HTTPClientResponseDelegate { typealias Response = Void - init( - backpressureEventLoop: EventLoop? = nil, - stateDidChangeCallback: ((State) -> Void)? = nil - ) { + init(backpressureEventLoop: EventLoop? = nil) { self.backpressureEventLoop = backpressureEventLoop - self.stateDidChangeCallback = stateDidChangeCallback } var backpressureEventLoop: EventLoop? - var stateDidChangeCallback: ((State) -> Void)? enum State { case idle @@ -68,11 +63,7 @@ class TestHTTPDelegate: HTTPClientResponseDelegate { case error(Error) } - var state = State.idle { - didSet { - self.stateDidChangeCallback?(self.state) - } - } + var state = State.idle func didReceiveHead(task: HTTPClient.Task, _ head: HTTPResponseHead) -> EventLoopFuture { self.state = .head(head) From c82a3a5fdc8cfbddd21219afa075559a0677c7b0 Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Thu, 11 Nov 2021 11:07:35 +0100 Subject: [PATCH 7/9] only send header once --- Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index a9ab7772e..04c556ecf 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -379,7 +379,10 @@ private final class SendHeaderAndWaitChannelHandler: ChannelInboundHandler { typealias OutboundOut = HTTPServerResponsePart func channelRead(context: ChannelHandlerContext, data: NIOAny) { - context.writeAndFlush(wrapOutboundOut(.head(HTTPResponseHead( + let requestPart = self.unwrapInboundIn(data) + guard case .end = requestPart else { return } + + context.writeAndFlush(self.wrapOutboundOut(.head(HTTPResponseHead( version: HTTPVersion(major: 1, minor: 1), status: .ok )) From b5391fd2ba23d18cebf58552e1a3f90b928ec7d3 Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Thu, 11 Nov 2021 11:23:36 +0100 Subject: [PATCH 8/9] send header as early as posible --- Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 04c556ecf..1d19b8ddc 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -380,7 +380,7 @@ private final class SendHeaderAndWaitChannelHandler: ChannelInboundHandler { func channelRead(context: ChannelHandlerContext, data: NIOAny) { let requestPart = self.unwrapInboundIn(data) - guard case .end = requestPart else { return } + guard case .head = requestPart else { return } context.writeAndFlush(self.wrapOutboundOut(.head(HTTPResponseHead( version: HTTPVersion(major: 1, minor: 1), From 3401a165c3cd9b20e8961761db59eef5239a6a81 Mon Sep 17 00:00:00 2001 From: David Nadoba Date: Thu, 11 Nov 2021 11:33:13 +0100 Subject: [PATCH 9/9] use switch --- .../AsyncHTTPClientTests/HTTP2ClientTests.swift | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift index 1d19b8ddc..34076f75c 100644 --- a/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTP2ClientTests.swift @@ -380,12 +380,15 @@ private final class SendHeaderAndWaitChannelHandler: ChannelInboundHandler { func channelRead(context: ChannelHandlerContext, data: NIOAny) { let requestPart = self.unwrapInboundIn(data) - guard case .head = requestPart else { return } - - context.writeAndFlush(self.wrapOutboundOut(.head(HTTPResponseHead( - version: HTTPVersion(major: 1, minor: 1), - status: .ok - )) - ), promise: nil) + switch requestPart { + case .head: + context.writeAndFlush(self.wrapOutboundOut(.head(HTTPResponseHead( + version: HTTPVersion(major: 1, minor: 1), + status: .ok + )) + ), promise: nil) + case .body, .end: + return + } } }