Repository: vapor/vapor
File: Tests/AsyncTests/AsyncRequestTests.swift (Swift)

Code before:
        XCTAssertEqual(body.string, testValue)
    }
}

// This was taken from AsyncHTTPClients's AsyncRequestTests.swift code.
// The license for the original work is reproduced below. See NOTICES.txt for
// more.

//===----------------------------------------------------------------------===//
//
// This source file is part of the AsyncHTTPClient open source project
//
// Copyright (c) 2022 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
//
//===----------------------------------------------------------------------===//

@usableFromInline
struct AsyncLazySequence<Base: Sequence>: AsyncSequence {
    @usableFromInline typealias Element = Base.Element
    @usableFromInline struct AsyncIterator: AsyncIteratorProtocol {
        @usableFromInline var iterator: Base.Iterator
        @inlinable init(iterator: Base.Iterator) {
            self.iterator = iterator
        }

        @inlinable mutating func next() async throws -> Base.Element? {
            self.iterator.next()
        }
    }

    @usableFromInline var base: Base

    @inlinable init(base: Base) {
        self.base = base
    }

    @inlinable func makeAsyncIterator() -> AsyncIterator {
        .init(iterator: self.base.makeIterator())
    }
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
extension AsyncLazySequence: Sendable where Base: Sendable {}
@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
extension AsyncLazySequence.AsyncIterator: Sendable where Base.Iterator: Sendable {}

Review comment (suggestion):
We could just use `AsyncAlgorithms`'s `var async`, but not sure if it's fine to add a dependency to that just for this test.
That also takes Task cancellation into account which we don't want, but this is in tests and that's not going to happen anyway.
https://github.com/apple/swift-async-algorithms/blob/5bbdcc1d37990cca524852c5a3924c02fa2da983/Sources/AsyncAlgorithms/AsyncSyncSequence.swift#L46C5-L46C5

Code after:
        
        let response: HTTPClientResponse = try await app.http.client.shared.execute(request, timeout: .seconds(5))
        XCTAssertEqual(response.status, .ok)
        let body = try await response.body.collect(upTo: 1024 * 1024)
        XCTAssertEqual(body.string, testValue)
    }
    
    func testStreamingRequestBodyCleansUp() async throws {
        app.http.server.configuration.hostname = "127.0.0.1"
        app.http.server.configuration.port = 0
        
        let bytesTheServerRead = ManagedAtomic<Int>(0)
        
        app.on(.POST, "hello", body: .stream) { req async throws -> Response in
            var bodyIterator = req.body.makeAsyncIterator()
            let firstChunk = try await bodyIterator.next()
            bytesTheServerRead.wrappingIncrement(by: firstChunk?.readableBytes ?? 0, ordering: .relaxed)
            throw Abort(.internalServerError)
        }
        
        app.environment.arguments = ["serve"]
        XCTAssertNoThrow(try app.start())
        
        XCTAssertNotNil(app.http.server.shared.localAddress)
        guard let localAddress = app.http.server.shared.localAddress,
              let ip = localAddress.ipAddress,
              let port = localAddress.port else {
            XCTFail("couldn't get ip/port from \(app.http.server.shared.localAddress.debugDescription)")
            return
        }
        
        var oneMBBB = ByteBuffer(repeating: 0x41, count: 1024 * 1024)
        let oneMB = try XCTUnwrap(oneMBBB.readData(length: oneMBBB.readableBytes))
        var request = HTTPClientRequest(url: "http://\(ip):\(port)/hello")
        request.method = .POST
        request.body = .stream(oneMB.async, length: .known(oneMB.count))
        let response = try await app.http.client.shared.execute(request, timeout: .seconds(5))
        
        XCTAssertGreaterThan(bytesTheServerRead.load(ordering: .relaxed), 0)
        XCTAssertEqual(response.status, .internalServerError)
    }
    
    // TODO: Re-enable once it reliably works and doesn't cause issues with trying to shut the application down
    // This may require some work in Vapor
    func _testRequestBodyBackpressureWorksWithAsyncStreaming() async throws {
        app.http.server.configuration.hostname = "127.0.0.1"
        app.http.server.configuration.port = 0
        
        let numberOfTimesTheServerGotOfferedBytes = ManagedAtomic<Int>(0)
        let bytesTheServerSaw = ManagedAtomic<Int>(0)
        let bytesTheClientSent = ManagedAtomic<Int>(0)