save progress

fix deprecation warnings for `Task.sleep(_:)`

rename AsyncRequest to HTTPClientRequest

rename AsyncResponse to HTTPClientResponse

fix merge conflicts

adopt new validation method

fix compilation

Author: dnadoba

Sources/AsyncHTTPClient/AsyncAwait/AsyncRequest+Validation.swift

@@ -16,7 +16,7 @@ import struct Foundation.URL
 import NIOHTTP1
 
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
-extension AsyncRequest {
+extension HTTPClientRequest {
 
     struct ValidationResult {
         let requestFramingMetadata: RequestFramingMetadata
@@ -45,10 +45,11 @@ extension AsyncRequest {
             tlsConfiguration: nil
         )
 
-        try self.headers.validateFieldNames()
-        
         var head = HTTPRequestHead(version: .http1_1, method: self.method, uri: url.uri, headers: self.headers)
 
+        let bodyLength = RequestBodyLength(self.body)
+        let metadata = try head.headers.validateAndSetTransportFraming(method: self.method, bodyLength: bodyLength)
+        
         // if no host header was set, let's pick
         if !head.headers.contains(name: "host") {
             guard let urlHost = url.host else {
@@ -57,67 +58,22 @@ extension AsyncRequest {
             head.headers.add(name: "host", value: urlHost)
         }
 
-        let encodings = head.headers[canonicalForm: "Transfer-Encoding"].map { $0.lowercased() }
-        if encodings.contains("identity") {
-            throw HTTPClientError.identityCodingIncorrectlyPresent
-        }
-        
-        head.headers.remove(name: "Transfer-Encoding")
-        
-        guard let body = self.body else {
-            head.headers.remove(name: "Content-Length")
-            // if we don't have a body we might not need to send the Content-Length field
-            // https://tools.ietf.org/html/rfc7230#section-3.3.2
-            switch method {
-            case .GET, .HEAD, .DELETE, .CONNECT, .TRACE:
-                // A user agent SHOULD NOT send a Content-Length header field when the request
-                // message does not contain a payload body and the method semantics do not
-                // anticipate such a body.
-                return ValidationResult(
-                    requestFramingMetadata: .init(connectionClose: !head.isKeepAlive, body: .none),
-                    poolKey: poolKey,
-                    head: head
-                )
-            default:
-                // A user agent SHOULD send a Content-Length in a request message when
-                // no Transfer-Encoding is sent and the request method defines a meaning
-                // for an enclosed payload body.
-                head.headers.add(name: "Content-Length", value: "0")
-                return ValidationResult(
-                    requestFramingMetadata: .init(connectionClose: !head.isKeepAlive, body: .none),
-                    poolKey: poolKey,
-                    head: head
-                )
-            }
-        }
-        
-        if case .TRACE = method {
-            // A client MUST NOT send a message body in a TRACE request.
-            // https://tools.ietf.org/html/rfc7230#section-4.3.8
-            throw HTTPClientError.traceRequestWithBody
-        }
+        return .init(requestFramingMetadata: metadata, poolKey: poolKey, head: head)
+    }
+}
 
-        guard (encodings.lazy.filter { $0 == "chunked" }.count <= 1) else {
-            throw HTTPClientError.chunkedSpecifiedMultipleTimes
+extension RequestBodyLength {
+    @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+    init(_ body: HTTPClientRequest.Body?) {
+        switch body?.mode {
+        case .none:
+            self = .fixed(length: 0)
+        case .byteBuffer(let buffer):
+            self = .fixed(length: buffer.readableBytes)
+        case .sequence(nil, _), .asyncSequence(nil, _):
+            self = .dynamic
+        case .sequence(.some(let length), _), .asyncSequence(.some(let length), _):
+            self = .fixed(length: length)
         }
-
-//        if encodings.isEmpty {
-//            switch self.body {
-//            case .some(.byteBuffer(let byteBuffer)):
-//                head.headers.add(name: "content-length", value: "\(byteBuffer.readableBytes)")
-//            case .some(bytes(let sequence)):
-//                // if we have a content length header, we assume this was set correctly
-//                if head.headers.contains(name: "content-length") {
-//                    
-//                } else {
-//                    head.headers.add(name: "transfer-encoding", value: "chunked")
-//                }
-//                
-//            }
-//            
-//            
-//        }
-        
-        preconditionFailure()
     }
 }

Sources/AsyncHTTPClient/AsyncAwait/AsyncRequestBag+StateMachine.swift

@@ -23,15 +23,15 @@ extension AsyncRequestBag {
         struct ExecutionContext {
             let executor: HTTPRequestExecutor
             let allocator: ByteBufferAllocator
-            let continuation: UnsafeContinuation<AsyncResponse, Error>
+            let continuation: UnsafeContinuation<HTTPClientResponse, Error>
         }
 
         private enum State {
             case initialized
-            case waiting(UnsafeContinuation<AsyncResponse, Error>)
-            case queued(UnsafeContinuation<AsyncResponse, Error>, HTTPRequestScheduler)
+            case waiting(UnsafeContinuation<HTTPClientResponse, Error>)
+            case queued(UnsafeContinuation<HTTPClientResponse, Error>, HTTPRequestScheduler)
             case executing(ExecutionContext, RequestStreamState, ResponseStreamState)
-            case finished(error: Error?, AsyncResponse.Body.IteratorStream.ID?)
+            case finished(error: Error?, HTTPClientResponse.Body.IteratorStream.ID?)
         }
 
         fileprivate enum RequestStreamState {
@@ -50,9 +50,9 @@ extension AsyncRequestBag {
 
             case initialized
             case waitingForStream(CircularBuffer<ByteBuffer>, next: Next)
-            case buffering(AsyncResponse.Body.IteratorStream.ID, CircularBuffer<ByteBuffer>, next: Next)
-            case waitingForRemote(AsyncResponse.Body.IteratorStream.ID, UnsafeContinuation<ByteBuffer?, Error>)
-            case finished(AsyncResponse.Body.IteratorStream.ID, UnsafeContinuation<ByteBuffer?, Error>)
+            case buffering(HTTPClientResponse.Body.IteratorStream.ID, CircularBuffer<ByteBuffer>, next: Next)
+            case waitingForRemote(HTTPClientResponse.Body.IteratorStream.ID, UnsafeContinuation<ByteBuffer?, Error>)
+            case finished(HTTPClientResponse.Body.IteratorStream.ID, UnsafeContinuation<ByteBuffer?, Error>)
         }
 
         private var state: State
@@ -61,7 +61,7 @@ extension AsyncRequestBag {
             self.state = .initialized
         }
 
-        mutating func registerContinuation(_ continuation: UnsafeContinuation<AsyncResponse, Error>) {
+        mutating func registerContinuation(_ continuation: UnsafeContinuation<HTTPClientResponse, Error>) {
             guard case .initialized = self.state else {
                 preconditionFailure("Invalid state: \(self.state)")
             }
@@ -95,7 +95,7 @@ extension AsyncRequestBag {
 
         enum FailAction {
             case none
-            case failContinuation(UnsafeContinuation<AsyncResponse, Error>, Error, HTTPRequestScheduler?, HTTPRequestExecutor?)
+            case failContinuation(UnsafeContinuation<HTTPClientResponse, Error>, Error, HTTPRequestScheduler?, HTTPRequestExecutor?)
             case failResponseStream(UnsafeContinuation<ByteBuffer?, Error>, Error, HTTPRequestExecutor)
         }
 
@@ -333,7 +333,7 @@ extension AsyncRequestBag {
         // MARK: - Response -
 
         enum ReceiveResponseHeadAction {
-            case succeedResponseHead(HTTPResponseHead, UnsafeContinuation<AsyncResponse, Error>)
+            case succeedResponseHead(HTTPResponseHead, UnsafeContinuation<HTTPClientResponse, Error>)
             case none
         }
 
@@ -421,7 +421,7 @@ extension AsyncRequestBag {
         struct TriedToRegisteredASecondConsumer: Error {}
 
         mutating func consumeNextResponsePart(
-            streamID: AsyncResponse.Body.IteratorStream.ID,
+            streamID: HTTPClientResponse.Body.IteratorStream.ID,
             continuation: UnsafeContinuation<ByteBuffer?, Error>
         ) -> ConsumeAction {
             switch self.state {

Sources/AsyncHTTPClient/AsyncAwait/AsyncRequestBag.swift

@@ -20,7 +20,7 @@ import NIOHTTP1
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
 actor AsyncRequestBag {
     // TODO: We should drop the request after sending to free up resource ASAP
-    let request: AsyncRequest
+    let request: HTTPClientRequest
 
     let logger: Logger
 
@@ -35,36 +35,23 @@ actor AsyncRequestBag {
     private var state: StateMachine = .init()
     private var isCancelled = false
 
-    init(request: AsyncRequest,
-         requestOptions: RequestOptions,
-         logger: Logger,
-         connectionDeadline: NIODeadline,
-         preferredEventLoop: EventLoop) {
+    init(
+        request: HTTPClientRequest,
+        requestOptions: RequestOptions,
+        logger: Logger,
+        connectionDeadline: NIODeadline,
+        preferredEventLoop: EventLoop
+    ) throws {
         self.request = request
         self.requestOptions = requestOptions
         self.logger = logger
         self.connectionDeadline = connectionDeadline
         self.preferredEventLoop = preferredEventLoop
 
-        self.requestHead = HTTPRequestHead(
-            version: .http1_1,
-            method: request.method,
-            uri: request.url,
-            headers: request.headers
-        )
-
-        switch request.body?.mode {
-        case .byteBuffer(let byteBuffer):
-            self.requestFramingMetadata = .init(connectionClose: false, body: .fixedSize(byteBuffer.readableBytes))
-        case .sequence:
-            self.requestFramingMetadata = .init(connectionClose: false, body: .stream)
-        case .asyncSequence:
-            self.requestFramingMetadata = .init(connectionClose: false, body: .stream)
-        case .none:
-            self.requestFramingMetadata = .init(connectionClose: false, body: .none)
-        }
-        
-        self.poolKey = ConnectionPool.Key(scheme: .http, host: "localhost", port: 123, unixPath: "", tlsConfiguration: nil)
+        let validatedRequest = try request.validate()
+        self.poolKey = validatedRequest.poolKey
+        self.requestHead = validatedRequest.head
+        self.requestFramingMetadata = validatedRequest.requestFramingMetadata
     }
 
     nonisolated func cancel() {
@@ -73,7 +60,7 @@ actor AsyncRequestBag {
         }
     }
 
-    func result() async throws -> AsyncResponse {
+    func result() async throws -> HTTPClientResponse {
         try await withUnsafeThrowingContinuation { continuation in
             self.state.registerContinuation(continuation)
         }
@@ -120,7 +107,7 @@ actor AsyncRequestBag {
             break
         case .resumeStream(let allocator):
             switch self.request.body?.mode {
-            case .asyncSequence(let next):
+            case .asyncSequence(_, let next):
                 // it is safe to call this async here. it dispatches...
                 await self.writeRequestStream(allocator, next: next)
 
@@ -138,7 +125,7 @@ actor AsyncRequestBag {
             case .none:
                 break
 
-            case .sequence(let create):
+            case .sequence(_, let create):
                 do {
                     let byteBuffer = try create(allocator) // <--- only throw point
 
@@ -176,7 +163,7 @@ actor AsyncRequestBag {
         case .none:
             break
         case .succeedResponseHead(let head, let continuation):
-            let asyncResponse = AsyncResponse(
+            let asyncResponse = HTTPClientResponse(
                 bag: self,
                 version: head.version,
                 status: head.status,
@@ -317,7 +304,7 @@ extension AsyncRequestBag: HTTPExecutableRequest {
 
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
 extension AsyncRequestBag {
-    func nextResponsePart(streamID: AsyncResponse.Body.IteratorStream.ID) async throws -> ByteBuffer? {
+    func nextResponsePart(streamID: HTTPClientResponse.Body.IteratorStream.ID) async throws -> ByteBuffer? {
         try await withUnsafeThrowingContinuation { continuation in
             switch self.state.consumeNextResponsePart(streamID: streamID, continuation: continuation) {
             case .succeedContinuation(let continuation, let result):
@@ -330,9 +317,9 @@ extension AsyncRequestBag {
         }
     }
 
-    func cancelResponseStream0(streamID: AsyncResponse.Body.IteratorStream.ID) {}
+    func cancelResponseStream0(streamID: HTTPClientResponse.Body.IteratorStream.ID) {}
 
-    nonisolated func cancelResponseStream(streamID: AsyncResponse.Body.IteratorStream.ID) {
+    nonisolated func cancelResponseStream(streamID: HTTPClientResponse.Body.IteratorStream.ID) {
         Task.detached {
             await self.cancelResponseStream0(streamID: streamID)
         }

Sources/AsyncHTTPClient/AsyncAwait/AsyncRequest.swift renamed to Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest.swift

@@ -18,13 +18,13 @@ import NIO
 import NIOHTTP1
 
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
-// HTTPClient.AsyncRequest
-struct AsyncRequest {
+// HTTPClient.HTTPClientRequest
+struct HTTPClientRequest {
     public struct Body {
         internal enum Mode {
-            case asyncSequence((ByteBufferAllocator) async throws -> (IOData?))
+            case asyncSequence(length: Int?, (ByteBufferAllocator) async throws -> (IOData?))
 //            case asyncSequenceFactory(() -> Mode) // typealias (ByteBufferAllocator) async throws -> IOData?
-            case sequence((ByteBufferAllocator) throws -> ByteBuffer)
+            case sequence(length: Int?, (ByteBufferAllocator) throws -> ByteBuffer)
             case byteBuffer(ByteBuffer)
         }
 
@@ -39,19 +39,29 @@ struct AsyncRequest {
         }
 
         static func bytes<S: Sequence>(_ sequence: S) -> Body where S.Element == UInt8 {
-            self.init(.asyncSequence { allocator in
+            self.init(.sequence(length: nil) { allocator in
                 if let buffer = sequence.withContiguousStorageIfAvailable({ allocator.buffer(bytes: $0) }) {
                     // fastpath
-                    return .byteBuffer(buffer)
+                    return buffer
+                }
+                // potentially really slow path
+                return allocator.buffer(bytes: sequence)
+            })
+        }
+        static func bytes<C: RandomAccessCollection>(_ collection: C) -> Body where C.Element == UInt8 {
+            self.init(.sequence(length: collection.count) { allocator in
+                if let buffer = collection.withContiguousStorageIfAvailable({ allocator.buffer(bytes: $0) }) {
+                    // fastpath
+                    return buffer
                 }
                 // potentially really slow path
-                return .byteBuffer(allocator.buffer(bytes: sequence))
+                return allocator.buffer(bytes: collection)
             })
         }
 
         static func stream<S: AsyncSequence>(_ sequence: S) -> Body where S.Element == ByteBuffer {
             var iterator = sequence.makeAsyncIterator()
-            let body = self.init(.asyncSequence { _ -> IOData? in
+            let body = self.init(.asyncSequence(length: nil) { _ -> IOData? in
                 if let byteBuffer = try await iterator.next() {
                     return .byteBuffer(byteBuffer)
                 }
@@ -62,7 +72,7 @@ struct AsyncRequest {
 
         static func stream<S: AsyncSequence>(_ sequence: S) -> Body where S.Element == FileRegion {
             var iterator = sequence.makeAsyncIterator()
-            let body = self.init(.asyncSequence { _ in
+            let body = self.init(.asyncSequence(length: nil) { _ in
                 if let fileRegion = try await iterator.next() {
                     return .fileRegion(fileRegion)
                 }
@@ -73,7 +83,7 @@ struct AsyncRequest {
 
         static func stream<S: AsyncSequence>(_ sequence: S) -> Body where S.Element == UInt8 {
             var iterator = sequence.makeAsyncIterator()
-            let body = self.init(.asyncSequence { allocator -> IOData? in
+            let body = self.init(.asyncSequence(length: nil) { allocator -> IOData? in
                 var buffer = allocator.buffer(capacity: 1024) // TODO: Magic number
                 while buffer.writableBytes > 0, let byte = try await iterator.next() {
                     buffer.writeInteger(byte)
@@ -102,7 +112,7 @@ struct AsyncRequest {
 }
 
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
-public struct AsyncResponse {
+public struct HTTPClientResponse {
     public var version: HTTPVersion
     public var status: HTTPResponseStatus
     public var headers: HTTPHeaders
@@ -130,7 +140,7 @@ public struct AsyncResponse {
 }
 
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
-extension AsyncResponse.Body: AsyncSequence {
+extension HTTPClientResponse.Body: AsyncSequence {
     public typealias Element = ByteBuffer
     public typealias AsyncIterator = Iterator
 
@@ -180,18 +190,18 @@ extension AsyncResponse.Body: AsyncSequence {
 
 @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
 extension HTTPClient {
-    func execute(_ request: AsyncRequest, deadline: NIODeadline, logger: Logger) async throws -> AsyncResponse {
-        let bag = AsyncRequestBag(
+    func execute(_ request: HTTPClientRequest, deadline: NIODeadline, logger: Logger) async throws -> HTTPClientResponse {
+        let bag = try AsyncRequestBag(
             request: request,
-            requestOptions: .init(idleReadTimeout: nil, ignoreUncleanSSLShutdown: false),
+            requestOptions: .init(idleReadTimeout: nil),
             logger: logger,
             connectionDeadline: .now() + .seconds(10),
             preferredEventLoop: self.eventLoopGroup.next()
         )
 
         return try await withTaskCancellationHandler {
             bag.cancel()
-        } operation: { () -> AsyncResponse in
+        } operation: { () -> HTTPClientResponse in
             // first register the completion
             async let result = bag.result()