Skip to content

Commit 4603a80

Browse files
guoye-zhangguoye-zhang
authored
Add trailers and request body streaming experimental SPIs (#911)
Adding the SPIs necessary to support the AHC backend in https://github.com/apple/swift-http-api-proposal
1 parent 7744c2a commit 4603a80

8 files changed

Lines changed: 167 additions & 38 deletions

File tree

‎Package.swift‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,16 @@ let package = Package(
3434
products: [
3535
.library(name: "AsyncHTTPClient", targets: ["AsyncHTTPClient"])
3636
],
37+
traits: [
38+
.default(enabledTraits: []),
39+
.trait(
40+
name: "UnstableHTTPAPIsSupport",
41+
description: """
42+
Enables source unstable SPIs to support the swift-http-api-proposal
43+
AHC backend. Do not rely on this module in API stable packages.
44+
"""
45+
),
46+
],
3747
dependencies: [
3848
.package(url: "https://github.com/apple/swift-nio.git", from: "2.100.0"),
3949
.package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.30.0"),

‎Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest+Prepared.swift‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ extension HTTPClientRequest {
3838
makeCompleteBody: @Sendable (ByteBufferAllocator) -> ByteBuffer
3939
)
4040
case byteBuffer(ByteBuffer)
41+
#if UnstableHTTPAPIsSupport
42+
case httpClientRequestBody(RequestBodyLength, HTTPClientRequest.Body.RequestWriterContinuation)
43+
#endif
4144
}
4245

4346
var url: URL
@@ -111,6 +114,10 @@ extension HTTPClientRequest.Prepared.Body {
111114
)
112115
case .byteBuffer(let byteBuffer):
113116
self = .byteBuffer(byteBuffer)
117+
#if UnstableHTTPAPIsSupport
118+
case .httpClientRequestBody(let length, let requestBody):
119+
self = .httpClientRequestBody(length, requestBody)
120+
#endif
114121
}
115122
}
116123
}
@@ -125,6 +132,10 @@ extension RequestBodyLength {
125132
self = .known(Int64(buffer.readableBytes))
126133
case .sequence(let length, _, _), .asyncSequence(let length, _):
127134
self = length
135+
#if UnstableHTTPAPIsSupport
136+
case .httpClientRequestBody(let length, _):
137+
self = length
138+
#endif
128139
}
129140
}
130141
}

‎Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest.swift‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,13 @@ extension HTTPClientRequest {
100100
makeCompleteBody: @Sendable (ByteBufferAllocator) -> ByteBuffer
101101
)
102102
case byteBuffer(ByteBuffer)
103+
104+
#if UnstableHTTPAPIsSupport
105+
case httpClientRequestBody(
106+
length: RequestBodyLength,
107+
startUpload: RequestWriterContinuation
108+
)
109+
#endif
103110
}
104111

105112
@usableFromInline
@@ -109,6 +116,44 @@ extension HTTPClientRequest {
109116
internal init(_ mode: Mode) {
110117
self.mode = mode
111118
}
119+
120+
#if UnstableHTTPAPIsSupport
121+
public init(length: Int64?, startUpload: AsyncStream<RequestWriter>.Continuation) {
122+
let length = length.map { RequestBodyLength.known($0) } ?? .unknown
123+
self.init(
124+
.httpClientRequestBody(
125+
length: length,
126+
startUpload: RequestWriterContinuation(continuation: startUpload)
127+
)
128+
)
129+
}
130+
131+
@usableFromInline
132+
struct RequestWriterContinuation: Sendable {
133+
var continuation: AsyncStream<RequestWriter>.Continuation
134+
}
135+
136+
@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
137+
public struct RequestWriter: Sendable {
138+
@usableFromInline
139+
let transaction: Transaction
140+
141+
@inlinable
142+
public func writeRequestBodyPart(_ buffer: ByteBuffer) async throws {
143+
try await self.transaction.writeRequestBodyPart(buffer)
144+
}
145+
146+
@inlinable
147+
public func requestBodyStreamFinished(trailers: HTTPHeaders?) {
148+
self.transaction.requestBodyStreamFinished(trailers: trailers)
149+
}
150+
151+
@inlinable
152+
public func fail(_ error: any Error) {
153+
self.transaction.fail(error)
154+
}
155+
}
156+
#endif
112157
}
113158
}
114159

@@ -353,6 +398,9 @@ extension Optional where Wrapped == HTTPClientRequest.Body {
353398
case .byteBuffer: return true
354399
case .sequence(_, let canBeConsumedMultipleTimes, _): return canBeConsumedMultipleTimes
355400
case .asyncSequence: return false
401+
#if UnstableHTTPAPIsSupport
402+
case .httpClientRequestBody: return false // TODO: I think this should be TRUE
403+
#endif
356404
}
357405
}
358406
}
@@ -393,6 +441,10 @@ extension HTTPClientRequest.Body: AsyncSequence {
393441
return .init(storage: .byteBuffer(makeCompleteBody(AsyncIterator.allocator)))
394442
case .byteBuffer(let byteBuffer):
395443
return .init(storage: .byteBuffer(byteBuffer))
444+
#if UnstableHTTPAPIsSupport
445+
case .httpClientRequestBody:
446+
fatalError("Unimplemented")
447+
#endif
396448
}
397449
}
398450
}

‎Sources/AsyncHTTPClient/AsyncAwait/HTTPClientResponse.swift‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@ public struct HTTPClientResponse: Sendable {
8282
version: HTTPVersion,
8383
status: HTTPResponseStatus,
8484
headers: HTTPHeaders,
85+
transaction: Transaction,
8586
body: TransactionBody,
8687
history: [HTTPClientRequestResponse]
8788
) {
@@ -92,6 +93,7 @@ public struct HTTPClientResponse: Sendable {
9293
body: .init(
9394
.transaction(
9495
body,
96+
transaction,
9597
expectedContentLength: HTTPClientResponse.expectedContentLength(
9698
requestMethod: requestMethod,
9799
headers: headers,
@@ -153,7 +155,7 @@ extension HTTPClientResponse {
153155
/// - Returns: the number of bytes collected over time
154156
@inlinable public func collect(upTo maxBytes: Int) async throws -> ByteBuffer {
155157
switch self.storage {
156-
case .transaction(_, let expectedContentLength):
158+
case .transaction(_, _, let expectedContentLength):
157159
if let contentLength = expectedContentLength {
158160
if contentLength > maxBytes {
159161
throw NIOTooManyBytesError(maxBytes: maxBytes)
@@ -170,6 +172,18 @@ extension HTTPClientResponse {
170172
}
171173
return try await collect(self, maxBytes: maxBytes)
172174
}
175+
176+
#if UnstableHTTPAPIsSupport
177+
public var trailers: HTTPHeaders? {
178+
switch self.storage {
179+
case .transaction(_, let transaction, _):
180+
return transaction.trailers
181+
182+
case .anyAsyncSequence:
183+
return nil
184+
}
185+
}
186+
#endif
173187
}
174188
}
175189

@@ -203,7 +217,7 @@ typealias TransactionBody = NIOThrowingAsyncSequenceProducer<
203217
@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
204218
extension HTTPClientResponse.Body {
205219
@usableFromInline enum Storage: Sendable {
206-
case transaction(TransactionBody, expectedContentLength: Int?)
220+
case transaction(TransactionBody, Transaction, expectedContentLength: Int?)
207221
case anyAsyncSequence(AnyAsyncSequence<ByteBuffer>)
208222
}
209223
}
@@ -214,7 +228,7 @@ extension HTTPClientResponse.Body.Storage: AsyncSequence {
214228

215229
@inlinable func makeAsyncIterator() -> AsyncIterator {
216230
switch self {
217-
case .transaction(let transaction, _):
231+
case .transaction(let transaction, _, _):
218232
return .transaction(transaction.makeAsyncIterator())
219233
case .anyAsyncSequence(let anyAsyncSequence):
220234
return .anyAsyncSequence(anyAsyncSequence.makeAsyncIterator())

‎Sources/AsyncHTTPClient/AsyncAwait/Transaction+StateMachine.swift‎

Lines changed: 37 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ extension Transaction {
3131
case queued(CheckedContinuation<HTTPClientResponse, Error>, HTTPRequestScheduler)
3232
case deadlineExceededWhileQueued(CheckedContinuation<HTTPClientResponse, Error>)
3333
case executing(ExecutionContext, RequestStreamState, ResponseStreamState)
34-
case finished(error: Error?)
34+
case finished(Result<HTTPHeaders?, any Error>)
3535
}
3636

3737
fileprivate enum RequestStreamState: Sendable {
@@ -47,7 +47,7 @@ extension Transaction {
4747
case waitingForResponseHead
4848
// streaming response body. Valid transitions to: finished.
4949
case streamingBody(TransactionBody.Source)
50-
case finished
50+
case finished(HTTPHeaders?)
5151
}
5252

5353
private var state: State
@@ -105,11 +105,11 @@ extension Transaction {
105105
mutating func fail(_ error: Error) -> FailAction {
106106
switch self.state {
107107
case .initialized(let continuation):
108-
self.state = .finished(error: error)
108+
self.state = .finished(.failure(error))
109109
return .failResponseHead(continuation, error, nil, nil, bodyStreamContinuation: nil)
110110

111111
case .queued(let continuation, let scheduler):
112-
self.state = .finished(error: error)
112+
self.state = .finished(.failure(error))
113113
return .failResponseHead(continuation, error, scheduler, nil, bodyStreamContinuation: nil)
114114
case .deadlineExceededWhileQueued(let continuation):
115115
let realError: Error = {
@@ -123,12 +123,12 @@ extension Transaction {
123123
}
124124
}()
125125

126-
self.state = .finished(error: realError)
126+
self.state = .finished(.failure(realError))
127127
return .failResponseHead(continuation, realError, nil, nil, bodyStreamContinuation: nil)
128128
case .executing(let context, let requestStreamState, .waitingForResponseHead):
129129
switch requestStreamState {
130130
case .paused(continuation: .some(let continuation)):
131-
self.state = .finished(error: error)
131+
self.state = .finished(.failure(error))
132132
return .failResponseHead(
133133
context.continuation,
134134
error,
@@ -138,7 +138,7 @@ extension Transaction {
138138
)
139139

140140
case .requestHeadSent, .endForwarded, .finished, .producing, .paused(continuation: .none):
141-
self.state = .finished(error: error)
141+
self.state = .finished(.failure(error))
142142
return .failResponseHead(
143143
context.continuation,
144144
error,
@@ -149,7 +149,7 @@ extension Transaction {
149149
}
150150

151151
case .executing(let context, let requestStreamState, .streamingBody(let source)):
152-
self.state = .finished(error: error)
152+
self.state = .finished(.failure(error))
153153
switch requestStreamState {
154154
case .paused(let bodyStreamContinuation):
155155
return .failResponseStream(
@@ -164,7 +164,7 @@ extension Transaction {
164164

165165
case .executing(let context, let requestStreamState, .finished):
166166
// an error occured after full response received, but before the full request was sent
167-
self.state = .finished(error: error)
167+
self.state = .finished(.failure(error))
168168
switch requestStreamState {
169169
case .paused(let bodyStreamContinuation):
170170
if let bodyStreamContinuation {
@@ -205,14 +205,14 @@ extension Transaction {
205205
return .none
206206
case .deadlineExceededWhileQueued(let continuation):
207207
let error = HTTPClientError.deadlineExceeded
208-
self.state = .finished(error: error)
208+
self.state = .finished(.failure(error))
209209
return .cancelAndFail(executor, continuation, with: error)
210210

211-
case .finished(error: .some):
211+
case .finished(.failure):
212212
return .cancel(executor)
213213

214214
case .executing,
215-
.finished(error: .none):
215+
.finished(.success):
216216
preconditionFailure("Invalid state: \(self.state)")
217217
}
218218
}
@@ -402,8 +402,8 @@ extension Transaction {
402402
assertionFailure("Invalid state: \(self.state)")
403403
return .failure(HTTPClientError.internalStateFailure())
404404

405-
case .executing(_, .endForwarded, .finished):
406-
self.state = .finished(error: nil)
405+
case .executing(_, .endForwarded, .finished(let trailers)):
406+
self.state = .finished(.success(trailers))
407407
return .none
408408

409409
case .executing(let context, .endForwarded, let responseState):
@@ -446,12 +446,12 @@ extension Transaction {
446446
self.state = .executing(context, requestState, .streamingBody(body.source))
447447
return .succeedResponseHead(body.sequence, context.continuation)
448448

449-
case .finished(error: .some):
449+
case .finished(.failure):
450450
// If the request failed before, we don't need to do anything in response to
451451
// receiving the response head.
452452
return .none
453453

454-
case .finished(error: .none):
454+
case .finished(.success):
455455
preconditionFailure("How can the request be finished without error, before receiving response head?")
456456
}
457457
}
@@ -511,7 +511,10 @@ extension Transaction {
511511
case none
512512
}
513513

514-
mutating func receiveResponseEnd(_ newChunks: CircularBuffer<ByteBuffer>?) -> ReceiveResponseEndAction {
514+
mutating func receiveResponseEnd(
515+
_ newChunks: CircularBuffer<ByteBuffer>?,
516+
trailers: HTTPHeaders?
517+
) -> ReceiveResponseEndAction {
515518
switch self.state {
516519
case .initialized,
517520
.queued,
@@ -524,9 +527,9 @@ extension Transaction {
524527
case .executing(let context, let requestState, .streamingBody(let source)):
525528
switch requestState {
526529
case .finished:
527-
self.state = .finished(error: nil)
530+
self.state = .finished(.success(trailers))
528531
case .paused, .producing, .requestHeadSent, .endForwarded:
529-
self.state = .executing(context, requestState, .finished)
532+
self.state = .executing(context, requestState, .finished(trailers))
530533
}
531534
return .finishResponseStream(source, finalBody: newChunks)
532535

@@ -540,6 +543,18 @@ extension Transaction {
540543
}
541544
}
542545

546+
var trailers: HTTPHeaders? {
547+
switch self.state {
548+
case .deadlineExceededWhileQueued, .initialized, .queued,
549+
.executing(_, _, .waitingForResponseHead),
550+
.executing(_, _, .streamingBody),
551+
.finished(.failure):
552+
return nil
553+
case .executing(_, _, .finished(let trailers)), .finished(.success(let trailers)):
554+
return trailers
555+
}
556+
}
557+
543558
mutating func httpResponseStreamTerminated() -> FailAction {
544559
switch self.state {
545560
case .executing(_, _, .finished), .finished:
@@ -565,7 +580,7 @@ extension Transaction {
565580
let error = HTTPClientError.deadlineExceeded
566581
switch self.state {
567582
case .initialized(let continuation):
568-
self.state = .finished(error: error)
583+
self.state = .finished(.failure(error))
569584
return .cancel(
570585
requestContinuation: continuation,
571586
scheduler: nil,
@@ -583,15 +598,15 @@ extension Transaction {
583598
case .executing(let context, let requestStreamState, .waitingForResponseHead):
584599
switch requestStreamState {
585600
case .paused(continuation: .some(let continuation)):
586-
self.state = .finished(error: error)
601+
self.state = .finished(.failure(error))
587602
return .cancel(
588603
requestContinuation: context.continuation,
589604
scheduler: nil,
590605
executor: context.executor,
591606
bodyStreamContinuation: continuation
592607
)
593608
case .requestHeadSent, .endForwarded, .finished, .producing, .paused(continuation: .none):
594-
self.state = .finished(error: error)
609+
self.state = .finished(.failure(error))
595610
return .cancel(
596611
requestContinuation: context.continuation,
597612
scheduler: nil,

0 commit comments

Comments
 (0)