Replace TransactionBody with NIOAsyncSequenceProducer (#677)

* save progress

* Replace `TransactionBody` with `NIOAsyncSequenceProducer`

* test

* revert unnesscary changes

* Add end-to-end test

which currently fails because of apple/swift-nio#2398

* soundness

* Use latest swift-nio release

* throw CancellationError on task cancelation

* Fix Swift 5.5 & 5.6

Author: dnadoba

Package.swift

@@ -21,7 +21,7 @@ let package = Package(
         .library(name: "AsyncHTTPClient", targets: ["AsyncHTTPClient"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.42.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.50.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.22.0"),
         .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.19.0"),
         .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.13.0"),

Package@swift-5.5.swift

@@ -21,7 +21,7 @@ let package = Package(
         .library(name: "AsyncHTTPClient", targets: ["AsyncHTTPClient"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.42.0"),
+        .package(url: "https://github.com/apple/swift-nio.git", from: "2.50.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.22.0"),
         .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.19.0"),
         .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.13.0"),

Sources/AsyncHTTPClient/AsyncAwait/AnyAsyncSequenceProucerDelete.swift

@@ -0,0 +1,37 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the AsyncHTTPClient open source project
+//
+// Copyright (c) 2023 Apple Inc. and the AsyncHTTPClient project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of AsyncHTTPClient project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+
+@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
+@usableFromInline
+struct AnyAsyncSequenceProducerDelegate: NIOAsyncSequenceProducerDelegate {
+    @usableFromInline
+    var delegate: NIOAsyncSequenceProducerDelegate
+
+    @inlinable
+    init<Delegate: NIOAsyncSequenceProducerDelegate>(_ delegate: Delegate) {
+        self.delegate = delegate
+    }
+
+    @inlinable
+    func produceMore() {
+        self.delegate.produceMore()
+    }
+
+    @inlinable
+    func didTerminate() {
+        self.delegate.didTerminate()
+    }
+}

Sources/AsyncHTTPClient/AsyncAwait/HTTPClientResponse.swift

@@ -45,14 +45,25 @@ public struct HTTPClientResponse: Sendable {
     }
 
     init(
-        bag: Transaction,
+        requestMethod: HTTPMethod,
         version: HTTPVersion,
         status: HTTPResponseStatus,
         headers: HTTPHeaders,
-        requestMethod: HTTPMethod
+        body: TransactionBody
     ) {
-        let contentLength = HTTPClientResponse.expectedContentLength(requestMethod: requestMethod, headers: headers, status: status)
-        self.init(version: version, status: status, headers: headers, body: .init(TransactionBody(bag, expectedContentLength: contentLength)))
+        self.init(
+            version: version,
+            status: status,
+            headers: headers,
+            body: .init(.transaction(
+                body,
+                expectedContentLength: HTTPClientResponse.expectedContentLength(
+                    requestMethod: requestMethod,
+                    headers: headers,
+                    status: status
+                )
+            ))
+        )
     }
 }
 
@@ -94,8 +105,8 @@ 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 transactionBody):
-                if let contentLength = transactionBody.expectedContentLength {
+            case .transaction(_, let expectedContentLength):
+                if let contentLength = expectedContentLength {
                     if contentLength > maxBytes {
                         throw NIOTooManyBytesError()
                     }
@@ -127,10 +138,19 @@ extension HTTPClientResponse {
     }
 }
 
+@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
+@usableFromInline
+typealias TransactionBody = NIOThrowingAsyncSequenceProducer<
+    ByteBuffer,
+    Error,
+    NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark,
+    AnyAsyncSequenceProducerDelegate
+>
+
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
 extension HTTPClientResponse.Body {
     @usableFromInline enum Storage: Sendable {
-        case transaction(TransactionBody)
+        case transaction(TransactionBody, expectedContentLength: Int?)
         case anyAsyncSequence(AnyAsyncSequence<ByteBuffer>)
     }
 }
@@ -141,7 +161,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())
@@ -172,10 +192,6 @@ extension HTTPClientResponse.Body.Storage.AsyncIterator: AsyncIteratorProtocol {
 
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
 extension HTTPClientResponse.Body {
-    init(_ body: TransactionBody) {
-        self.init(storage: .transaction(body))
-    }
-
     @inlinable init(_ storage: Storage) {
         self.storage = storage
     }