Add trailers and request body streaming experimental SPIs

Package.swift

@@ -34,6 +34,16 @@ let package = Package(
     products: [
         .library(name: "AsyncHTTPClient", targets: ["AsyncHTTPClient"])
     ],
+    traits: [
+        .default(enabledTraits: []),
+        .trait(
+            name: "UnstableHTTPAPIsSupport",
+            description: """
+                Enables source unstable SPIs to support the swift-http-api-proposal
+                AHC backend. Do not rely on this module in API stable packages.
+                """
+        ),
+    ],
     dependencies: [
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.100.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.30.0"),

Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest+Prepared.swift

@@ -38,6 +38,9 @@ extension HTTPClientRequest {
                 makeCompleteBody: @Sendable (ByteBufferAllocator) -> ByteBuffer
             )
             case byteBuffer(ByteBuffer)
+            #if UnstableHTTPAPIsSupport
+            case httpClientRequestBody(RequestBodyLength, HTTPClientRequest.Body.RequestWriterContinuation)
+            #endif
         }
 
         var url: URL
@@ -111,6 +114,10 @@ extension HTTPClientRequest.Prepared.Body {
             )
         case .byteBuffer(let byteBuffer):
             self = .byteBuffer(byteBuffer)
+        #if UnstableHTTPAPIsSupport
+        case .httpClientRequestBody(let length, let requestBody):
+            self = .httpClientRequestBody(length, requestBody)
+        #endif
         }
     }
 }
@@ -125,6 +132,10 @@ extension RequestBodyLength {
             self = .known(Int64(buffer.readableBytes))
         case .sequence(let length, _, _), .asyncSequence(let length, _):
             self = length
+        #if UnstableHTTPAPIsSupport
+        case .httpClientRequestBody(let length, _):
+            self = length
+        #endif
         }
     }
 }

Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest.swift

@@ -100,6 +100,13 @@ extension HTTPClientRequest {
                 makeCompleteBody: @Sendable (ByteBufferAllocator) -> ByteBuffer
             )
             case byteBuffer(ByteBuffer)
+
+            #if UnstableHTTPAPIsSupport
+            case httpClientRequestBody(
+                length: RequestBodyLength,
+                startUpload: RequestWriterContinuation
+            )
+            #endif
         }
 
         @usableFromInline
@@ -109,6 +116,44 @@ extension HTTPClientRequest {
         internal init(_ mode: Mode) {
             self.mode = mode
         }
+
+        #if UnstableHTTPAPIsSupport
+        public init(length: Int64?, startUpload: AsyncStream<RequestWriter>.Continuation) {
+            let length = length.map { RequestBodyLength.known($0) } ?? .unknown
+            self.init(
+                .httpClientRequestBody(
+                    length: length,
+                    startUpload: RequestWriterContinuation(continuation: startUpload)
+                )
+            )
+        }
+
+        @usableFromInline
+        struct RequestWriterContinuation: Sendable {
+            var continuation: AsyncStream<RequestWriter>.Continuation
+        }
+
+        @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
+        public struct RequestWriter: Sendable {
+            @usableFromInline
+            let transaction: Transaction
+
+            @inlinable
+            public func writeRequestBodyPart(_ buffer: ByteBuffer) async throws {
+                try await self.transaction.writeRequestBodyPart(buffer)
+            }
+
+            @inlinable
+            public func requestBodyStreamFinished(trailers: HTTPHeaders?) {
+                self.transaction.requestBodyStreamFinished(trailers: trailers)
+            }
+
+            @inlinable
+            public func fail(_ error: any Error) {
+                self.transaction.fail(error)
+            }
+        }
+        #endif
     }
 }
 
@@ -353,6 +398,9 @@ extension Optional where Wrapped == HTTPClientRequest.Body {
         case .byteBuffer: return true
         case .sequence(_, let canBeConsumedMultipleTimes, _): return canBeConsumedMultipleTimes
         case .asyncSequence: return false
+        #if UnstableHTTPAPIsSupport
+        case .httpClientRequestBody: return false  // TODO: I think this should be TRUE
+        #endif
         }
     }
 }
@@ -393,6 +441,10 @@ extension HTTPClientRequest.Body: AsyncSequence {
             return .init(storage: .byteBuffer(makeCompleteBody(AsyncIterator.allocator)))
         case .byteBuffer(let byteBuffer):
             return .init(storage: .byteBuffer(byteBuffer))
+        #if UnstableHTTPAPIsSupport
+        case .httpClientRequestBody:
+            fatalError("Unimplemented")
+        #endif
         }
     }
 }

Sources/AsyncHTTPClient/AsyncAwait/HTTPClientResponse.swift

@@ -82,6 +82,7 @@ public struct HTTPClientResponse: Sendable {
         version: HTTPVersion,
         status: HTTPResponseStatus,
         headers: HTTPHeaders,
+        transaction: Transaction,
         body: TransactionBody,
         history: [HTTPClientRequestResponse]
     ) {
@@ -92,6 +93,7 @@ public struct HTTPClientResponse: Sendable {
             body: .init(
                 .transaction(
                     body,
+                    transaction,
                     expectedContentLength: HTTPClientResponse.expectedContentLength(
                         requestMethod: requestMethod,
                         headers: headers,
@@ -153,7 +155,7 @@ extension HTTPClientResponse {
         /// - Returns: the number of bytes collected over time
         @inlinable public func collect(upTo maxBytes: Int) async throws -> ByteBuffer {
             switch self.storage {
-            case .transaction(_, let expectedContentLength):
+            case .transaction(_, _, let expectedContentLength):
                 if let contentLength = expectedContentLength {
                     if contentLength > maxBytes {
                         throw NIOTooManyBytesError(maxBytes: maxBytes)
@@ -170,6 +172,18 @@ extension HTTPClientResponse {
             }
             return try await collect(self, maxBytes: maxBytes)
         }
+
+        #if UnstableHTTPAPIsSupport
+        public var trailers: HTTPHeaders? {
+            switch self.storage {
+            case .transaction(_, let transaction, _):
+                return transaction.trailers
+
+            case .anyAsyncSequence:
+                return nil
+            }
+        }
+        #endif
     }
 }
 
@@ -203,7 +217,7 @@ typealias TransactionBody = NIOThrowingAsyncSequenceProducer<
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
 extension HTTPClientResponse.Body {
     @usableFromInline enum Storage: Sendable {
-        case transaction(TransactionBody, expectedContentLength: Int?)
+        case transaction(TransactionBody, Transaction, expectedContentLength: Int?)
         case anyAsyncSequence(AnyAsyncSequence<ByteBuffer>)
     }
 }
@@ -214,7 +228,7 @@ extension HTTPClientResponse.Body.Storage: AsyncSequence {
 
     @inlinable func makeAsyncIterator() -> AsyncIterator {
         switch self {
-        case .transaction(let transaction, _):
+        case .transaction(let transaction, _, _):
             return .transaction(transaction.makeAsyncIterator())
         case .anyAsyncSequence(let anyAsyncSequence):
             return .anyAsyncSequence(anyAsyncSequence.makeAsyncIterator())

Sources/AsyncHTTPClient/AsyncAwait/Transaction+StateMachine.swift

@@ -31,7 +31,7 @@ extension Transaction {
             case queued(CheckedContinuation<HTTPClientResponse, Error>, HTTPRequestScheduler)
             case deadlineExceededWhileQueued(CheckedContinuation<HTTPClientResponse, Error>)
             case executing(ExecutionContext, RequestStreamState, ResponseStreamState)
-            case finished(error: Error?)
+            case finished(Result<HTTPHeaders?, any Error>)
         }
 
         fileprivate enum RequestStreamState: Sendable {
@@ -47,7 +47,7 @@ extension Transaction {
             case waitingForResponseHead
             // streaming response body. Valid transitions to: finished.
             case streamingBody(TransactionBody.Source)
-            case finished
+            case finished(HTTPHeaders?)
         }
 
         private var state: State
@@ -105,11 +105,11 @@ extension Transaction {
         mutating func fail(_ error: Error) -> FailAction {
             switch self.state {
             case .initialized(let continuation):
-                self.state = .finished(error: error)
+                self.state = .finished(.failure(error))
                 return .failResponseHead(continuation, error, nil, nil, bodyStreamContinuation: nil)
 
             case .queued(let continuation, let scheduler):
-                self.state = .finished(error: error)
+                self.state = .finished(.failure(error))
                 return .failResponseHead(continuation, error, scheduler, nil, bodyStreamContinuation: nil)
             case .deadlineExceededWhileQueued(let continuation):
                 let realError: Error = {
@@ -123,12 +123,12 @@ extension Transaction {
                     }
                 }()
 
-                self.state = .finished(error: realError)
+                self.state = .finished(.failure(realError))
                 return .failResponseHead(continuation, realError, nil, nil, bodyStreamContinuation: nil)
             case .executing(let context, let requestStreamState, .waitingForResponseHead):
                 switch requestStreamState {
                 case .paused(continuation: .some(let continuation)):
-                    self.state = .finished(error: error)
+                    self.state = .finished(.failure(error))
                     return .failResponseHead(
                         context.continuation,
                         error,
@@ -138,7 +138,7 @@ extension Transaction {
                     )
 
                 case .requestHeadSent, .endForwarded, .finished, .producing, .paused(continuation: .none):
-                    self.state = .finished(error: error)
+                    self.state = .finished(.failure(error))
                     return .failResponseHead(
                         context.continuation,
                         error,
@@ -149,7 +149,7 @@ extension Transaction {
                 }
 
             case .executing(let context, let requestStreamState, .streamingBody(let source)):
-                self.state = .finished(error: error)
+                self.state = .finished(.failure(error))
                 switch requestStreamState {
                 case .paused(let bodyStreamContinuation):
                     return .failResponseStream(
@@ -164,7 +164,7 @@ extension Transaction {
 
             case .executing(let context, let requestStreamState, .finished):
                 // an error occured after full response received, but before the full request was sent
-                self.state = .finished(error: error)
+                self.state = .finished(.failure(error))
                 switch requestStreamState {
                 case .paused(let bodyStreamContinuation):
                     if let bodyStreamContinuation {
@@ -205,14 +205,14 @@ extension Transaction {
                 return .none
             case .deadlineExceededWhileQueued(let continuation):
                 let error = HTTPClientError.deadlineExceeded
-                self.state = .finished(error: error)
+                self.state = .finished(.failure(error))
                 return .cancelAndFail(executor, continuation, with: error)
 
-            case .finished(error: .some):
+            case .finished(.failure):
                 return .cancel(executor)
 
             case .executing,
-                .finished(error: .none):
+                .finished(.success):
                 preconditionFailure("Invalid state: \(self.state)")
             }
         }
@@ -402,8 +402,8 @@ extension Transaction {
                 assertionFailure("Invalid state: \(self.state)")
                 return .failure(HTTPClientError.internalStateFailure())
 
-            case .executing(_, .endForwarded, .finished):
-                self.state = .finished(error: nil)
+            case .executing(_, .endForwarded, .finished(let trailers)):
+                self.state = .finished(.success(trailers))
                 return .none
 
             case .executing(let context, .endForwarded, let responseState):
@@ -446,12 +446,12 @@ extension Transaction {
                 self.state = .executing(context, requestState, .streamingBody(body.source))
                 return .succeedResponseHead(body.sequence, context.continuation)
 
-            case .finished(error: .some):
+            case .finished(.failure):
                 // If the request failed before, we don't need to do anything in response to
                 // receiving the response head.
                 return .none
 
-            case .finished(error: .none):
+            case .finished(.success):
                 preconditionFailure("How can the request be finished without error, before receiving response head?")
             }
         }
@@ -511,7 +511,10 @@ extension Transaction {
             case none
         }
 
-        mutating func receiveResponseEnd(_ newChunks: CircularBuffer<ByteBuffer>?) -> ReceiveResponseEndAction {
+        mutating func receiveResponseEnd(
+            _ newChunks: CircularBuffer<ByteBuffer>?,
+            trailers: HTTPHeaders?
+        ) -> ReceiveResponseEndAction {
             switch self.state {
             case .initialized,
                 .queued,
@@ -524,9 +527,9 @@ extension Transaction {
             case .executing(let context, let requestState, .streamingBody(let source)):
                 switch requestState {
                 case .finished:
-                    self.state = .finished(error: nil)
+                    self.state = .finished(.success(trailers))
                 case .paused, .producing, .requestHeadSent, .endForwarded:
-                    self.state = .executing(context, requestState, .finished)
+                    self.state = .executing(context, requestState, .finished(trailers))
                 }
                 return .finishResponseStream(source, finalBody: newChunks)
 
@@ -540,6 +543,18 @@ extension Transaction {
             }
         }
 
+        var trailers: HTTPHeaders? {
+            switch self.state {
+            case .deadlineExceededWhileQueued, .initialized, .queued,
+                .executing(_, _, .waitingForResponseHead),
+                .executing(_, _, .streamingBody),
+                .finished(.failure):
+                return nil
+            case .executing(_, _, .finished(let trailers)), .finished(.success(let trailers)):
+                return trailers
+            }
+        }
+
         mutating func httpResponseStreamTerminated() -> FailAction {
             switch self.state {
             case .executing(_, _, .finished), .finished:
@@ -565,7 +580,7 @@ extension Transaction {
             let error = HTTPClientError.deadlineExceeded
             switch self.state {
             case .initialized(let continuation):
-                self.state = .finished(error: error)
+                self.state = .finished(.failure(error))
                 return .cancel(
                     requestContinuation: continuation,
                     scheduler: nil,
@@ -583,15 +598,15 @@ extension Transaction {
             case .executing(let context, let requestStreamState, .waitingForResponseHead):
                 switch requestStreamState {
                 case .paused(continuation: .some(let continuation)):
-                    self.state = .finished(error: error)
+                    self.state = .finished(.failure(error))
                     return .cancel(
                         requestContinuation: context.continuation,
                         scheduler: nil,
                         executor: context.executor,
                         bodyStreamContinuation: continuation
                     )
                 case .requestHeadSent, .endForwarded, .finished, .producing, .paused(continuation: .none):
-                    self.state = .finished(error: error)
+                    self.state = .finished(.failure(error))
                     return .cancel(
                         requestContinuation: context.continuation,
                         scheduler: nil,