Skip to content

Commit d928cc8

Browse files
fabianfettfabianfett
authored
[HTTPRequestStateMachine] Allow channelReadComplete at any time (#450)
1 parent a6ca288 commit d928cc8

File tree

5 files changed

+170
-1
lines changed

5 files changed

+170
-1
lines changed

‎Sources/AsyncHTTPClient/ConnectionPool/HTTPRequestStateMachine+Demand.swift‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,10 +92,21 @@ extension HTTPRequestStateMachine {
9292
return buffer
9393
}
9494

95+
// For all the following cases, please note:
96+
// Normally these code paths should never be hit. However there is one way to trigger
97+
// this:
98+
//
99+
// If the connection to a server is closed, NIO will forward all outstanding
100+
// `channelRead`s without waiting for a next `context.read` call. After all
101+
// `channelRead`s are delivered, we will also see a `channelReadComplete` call. After
102+
// this has happened, we know that we will get a channelInactive or further
103+
// `channelReads`. If the request ever gets to an `.end` all buffered data will be
104+
// forwarded to the user.
105+
95106
case .waitingForRead,
96107
.waitingForDemand,
97108
.waitingForReadOrDemand:
98-
preconditionFailure("How can we receive a body part, after a channelReadComplete, but no read has been forwarded yet. Invalid state: \(self.state)")
109+
return nil
99110

100111
case .modifying:
101112
preconditionFailure("Invalid state: \(self.state)")

‎Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests+XCTest.swift‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ extension HTTP1ClientChannelHandlerTests {
2929
("testWriteBackpressure", testWriteBackpressure),
3030
("testClientHandlerCancelsRequestIfWeWantToShutdown", testClientHandlerCancelsRequestIfWeWantToShutdown),
3131
("testIdleReadTimeout", testIdleReadTimeout),
32+
("testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand", testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand),
3233
]
3334
}
3435
}

‎Tests/AsyncHTTPClientTests/HTTP1ClientChannelHandlerTests.swift‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -286,6 +286,64 @@ class HTTP1ClientChannelHandlerTests: XCTestCase {
286286
XCTAssertEqual($0 as? HTTPClientError, .readTimeout)
287287
}
288288
}
289+
290+
func testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand() {
291+
let embedded = EmbeddedChannel()
292+
var maybeTestUtils: HTTP1TestTools?
293+
XCTAssertNoThrow(maybeTestUtils = try embedded.setupHTTP1Connection())
294+
guard let testUtils = maybeTestUtils else { return XCTFail("Expected connection setup works") }
295+
296+
var maybeRequest: HTTPClient.Request?
297+
XCTAssertNoThrow(maybeRequest = try HTTPClient.Request(url: "http://localhost/"))
298+
guard let request = maybeRequest else { return XCTFail("Expected to be able to create a request") }
299+
300+
let delegate = ResponseBackpressureDelegate(eventLoop: embedded.eventLoop)
301+
var maybeRequestBag: RequestBag<ResponseBackpressureDelegate>?
302+
XCTAssertNoThrow(maybeRequestBag = try RequestBag(
303+
request: request,
304+
eventLoopPreference: .delegate(on: embedded.eventLoop),
305+
task: .init(eventLoop: embedded.eventLoop, logger: testUtils.logger),
306+
redirectHandler: nil,
307+
connectionDeadline: .now() + .seconds(30),
308+
requestOptions: .forTests(),
309+
delegate: delegate
310+
))
311+
guard let requestBag = maybeRequestBag else { return XCTFail("Expected to be able to create a request bag") }
312+
313+
testUtils.connection.executeRequest(requestBag)
314+
315+
XCTAssertNoThrow(try embedded.receiveHeadAndVerify {
316+
XCTAssertEqual($0.method, .GET)
317+
XCTAssertEqual($0.uri, "/")
318+
XCTAssertEqual($0.headers.first(name: "host"), "localhost")
319+
})
320+
XCTAssertNoThrow(try embedded.receiveEnd())
321+
322+
let responseHead = HTTPResponseHead(version: .http1_1, status: .ok, headers: HTTPHeaders([("content-length", "50")]))
323+
324+
XCTAssertEqual(testUtils.readEventHandler.readHitCounter, 0)
325+
embedded.read()
326+
XCTAssertEqual(testUtils.readEventHandler.readHitCounter, 1)
327+
XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.head(responseHead)))
328+
329+
// not sending anything after the head should lead to request fail and connection close
330+
embedded.pipeline.fireChannelReadComplete()
331+
embedded.pipeline.read()
332+
XCTAssertEqual(testUtils.readEventHandler.readHitCounter, 2)
333+
334+
XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.body(ByteBuffer(string: "foo bar"))))
335+
embedded.pipeline.fireChannelReadComplete()
336+
// We miss a `embedded.pipeline.read()` here by purpose.
337+
XCTAssertEqual(testUtils.readEventHandler.readHitCounter, 2)
338+
339+
XCTAssertNoThrow(try embedded.writeInbound(HTTPClientResponsePart.body(ByteBuffer(string: "last bytes"))))
340+
embedded.pipeline.fireChannelReadComplete()
341+
embedded.pipeline.fireChannelInactive()
342+
343+
XCTAssertThrowsError(try requestBag.task.futureResult.wait()) {
344+
XCTAssertEqual($0 as? HTTPClientError, .remoteConnectionClosed)
345+
}
346+
}
289347
}
290348

291349
class TestBackpressureWriter {

‎Tests/AsyncHTTPClientTests/HTTPRequestStateMachineTests+XCTest.swift‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,10 @@ extension HTTPRequestStateMachineTests {
5656
("testFailHTTP1RequestWithoutContentLengthWithNIOSSLErrorUncleanShutdown", testFailHTTP1RequestWithoutContentLengthWithNIOSSLErrorUncleanShutdown),
5757
("testFailHTTP1RequestWithoutContentLengthWithNIOSSLErrorUncleanShutdownButIgnoreIt", testFailHTTP1RequestWithoutContentLengthWithNIOSSLErrorUncleanShutdownButIgnoreIt),
5858
("testFailHTTP1RequestWithContentLengthWithNIOSSLErrorUncleanShutdownButIgnoreIt", testFailHTTP1RequestWithContentLengthWithNIOSSLErrorUncleanShutdownButIgnoreIt),
59+
("testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand", testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand),
60+
("testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForRead", testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForRead),
61+
("testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForReadAndDemand", testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForReadAndDemand),
62+
("testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForReadAndDemandMultipleTimes", testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForReadAndDemandMultipleTimes),
5963
]
6064
}
6165
}

‎Tests/AsyncHTTPClientTests/HTTPRequestStateMachineTests.swift‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -557,6 +557,101 @@ class HTTPRequestStateMachineTests: XCTestCase {
557557
XCTAssertEqual(state.errorHappened(HTTPParserError.invalidEOFState), .failRequest(HTTPParserError.invalidEOFState, .close))
558558
XCTAssertEqual(state.channelInactive(), .wait)
559559
}
560+
561+
func testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForDemand() {
562+
var state = HTTPRequestStateMachine(isChannelWritable: true, ignoreUncleanSSLShutdown: false)
563+
let requestHead = HTTPRequestHead(version: .http1_1, method: .GET, uri: "/")
564+
let metadata = RequestFramingMetadata(connectionClose: false, body: .none)
565+
XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: false))
566+
567+
let responseHead = HTTPResponseHead(version: .http1_1, status: .ok, headers: ["Content-Length": "50"])
568+
let body = ByteBuffer(string: "foo bar")
569+
XCTAssertEqual(state.channelRead(.head(responseHead)), .forwardResponseHead(responseHead, pauseRequestBodyStream: false))
570+
XCTAssertEqual(state.demandMoreResponseBodyParts(), .wait)
571+
XCTAssertEqual(state.channelReadComplete(), .wait)
572+
XCTAssertEqual(state.read(), .read)
573+
XCTAssertEqual(state.channelRead(.body(body)), .wait)
574+
XCTAssertEqual(state.channelReadComplete(), .forwardResponseBodyParts([body]))
575+
XCTAssertEqual(state.read(), .wait)
576+
577+
XCTAssertEqual(state.channelRead(.body(ByteBuffer(string: " baz lightyear"))), .wait)
578+
XCTAssertEqual(state.channelReadComplete(), .wait)
579+
XCTAssertEqual(state.channelInactive(), .failRequest(HTTPClientError.remoteConnectionClosed, .none))
580+
}
581+
582+
func testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForRead() {
583+
var state = HTTPRequestStateMachine(isChannelWritable: true, ignoreUncleanSSLShutdown: false)
584+
let requestHead = HTTPRequestHead(version: .http1_1, method: .GET, uri: "/")
585+
let metadata = RequestFramingMetadata(connectionClose: false, body: .none)
586+
XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: false))
587+
588+
let responseHead = HTTPResponseHead(version: .http1_1, status: .ok, headers: ["Content-Length": "50"])
589+
let body = ByteBuffer(string: "foo bar")
590+
XCTAssertEqual(state.channelRead(.head(responseHead)), .forwardResponseHead(responseHead, pauseRequestBodyStream: false))
591+
XCTAssertEqual(state.demandMoreResponseBodyParts(), .wait)
592+
XCTAssertEqual(state.channelReadComplete(), .wait)
593+
XCTAssertEqual(state.read(), .read)
594+
XCTAssertEqual(state.channelRead(.body(body)), .wait)
595+
XCTAssertEqual(state.channelReadComplete(), .forwardResponseBodyParts([body]))
596+
XCTAssertEqual(state.demandMoreResponseBodyParts(), .wait)
597+
598+
XCTAssertEqual(state.channelRead(.body(ByteBuffer(string: " baz lightyear"))), .wait)
599+
XCTAssertEqual(state.channelReadComplete(), .wait)
600+
XCTAssertEqual(state.channelInactive(), .failRequest(HTTPClientError.remoteConnectionClosed, .none))
601+
}
602+
603+
func testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForReadAndDemand() {
604+
var state = HTTPRequestStateMachine(isChannelWritable: true, ignoreUncleanSSLShutdown: false)
605+
let requestHead = HTTPRequestHead(version: .http1_1, method: .GET, uri: "/")
606+
let metadata = RequestFramingMetadata(connectionClose: false, body: .none)
607+
XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: false))
608+
609+
let responseHead = HTTPResponseHead(version: .http1_1, status: .ok, headers: ["Content-Length": "50"])
610+
let body = ByteBuffer(string: "foo bar")
611+
XCTAssertEqual(state.channelRead(.head(responseHead)), .forwardResponseHead(responseHead, pauseRequestBodyStream: false))
612+
XCTAssertEqual(state.demandMoreResponseBodyParts(), .wait)
613+
XCTAssertEqual(state.channelReadComplete(), .wait)
614+
XCTAssertEqual(state.read(), .read)
615+
XCTAssertEqual(state.channelRead(.body(body)), .wait)
616+
XCTAssertEqual(state.channelReadComplete(), .forwardResponseBodyParts([body]))
617+
618+
XCTAssertEqual(state.channelRead(.body(ByteBuffer(string: " baz lightyear"))), .wait)
619+
XCTAssertEqual(state.channelReadComplete(), .wait)
620+
XCTAssertEqual(state.channelInactive(), .failRequest(HTTPClientError.remoteConnectionClosed, .none))
621+
}
622+
623+
func testFailHTTPRequestWithContentLengthBecauseOfChannelInactiveWaitingForReadAndDemandMultipleTimes() {
624+
var state = HTTPRequestStateMachine(isChannelWritable: true, ignoreUncleanSSLShutdown: false)
625+
let requestHead = HTTPRequestHead(version: .http1_1, method: .GET, uri: "/")
626+
let metadata = RequestFramingMetadata(connectionClose: false, body: .none)
627+
XCTAssertEqual(state.startRequest(head: requestHead, metadata: metadata), .sendRequestHead(requestHead, startBody: false))
628+
629+
let responseHead = HTTPResponseHead(version: .http1_1, status: .ok, headers: ["Content-Length": "50"])
630+
let body = ByteBuffer(string: "foo bar")
631+
XCTAssertEqual(state.channelRead(.head(responseHead)), .forwardResponseHead(responseHead, pauseRequestBodyStream: false))
632+
XCTAssertEqual(state.demandMoreResponseBodyParts(), .wait)
633+
XCTAssertEqual(state.channelReadComplete(), .wait)
634+
XCTAssertEqual(state.read(), .read)
635+
XCTAssertEqual(state.channelRead(.body(body)), .wait)
636+
XCTAssertEqual(state.channelReadComplete(), .forwardResponseBodyParts([body]))
637+
638+
let part1 = ByteBuffer(string: "baz lightyear")
639+
XCTAssertEqual(state.channelRead(.body(part1)), .wait)
640+
XCTAssertEqual(state.channelReadComplete(), .wait)
641+
642+
let part2 = ByteBuffer(string: "nearly last")
643+
XCTAssertEqual(state.channelRead(.body(part2)), .wait)
644+
XCTAssertEqual(state.channelReadComplete(), .wait)
645+
646+
let part3 = ByteBuffer(string: "final message")
647+
XCTAssertEqual(state.channelRead(.body(part3)), .wait)
648+
XCTAssertEqual(state.channelReadComplete(), .wait)
649+
650+
XCTAssertEqual(state.channelRead(.end(nil)), .succeedRequest(.close, [part1, part2, part3]))
651+
XCTAssertEqual(state.channelReadComplete(), .wait)
652+
653+
XCTAssertEqual(state.channelInactive(), .wait)
654+
}
560655
}
561656

562657
extension HTTPRequestStateMachine.Action: Equatable {

0 commit comments

Comments
 (0)