Code review

Author: fabianfett

Diff for: Sources/AsyncHTTPClient/ConnectionPool/HTTPExecutingRequest.swift

@@ -71,12 +71,9 @@ import NIOHTTP1
 ///
 /// The request execution will work as follows:
 ///
-///   1. The request executor will call `willExecuteRequest(_: HTTPRequestExecutor) -> Bool` on the
-///     request. The request is expected to return `true`, if it wants to be executed and false if
-///     it was cancelled or failed, before this checkpoint. If the request returns `true` it is
-///     expected to keep a reference to the `HTTPRequestExecutor` that was passed to the request
-///     for further communication. If the request returned false, the executor shall inform the
-///     scheduling mechanism that it is ready to execute the a new request.
+///   1. The request executor will call `willExecuteRequest(_: HTTPRequestExecutor)` on the
+///     request. The request is expected to keep a reference to the `HTTPRequestExecutor` that was
+///     passed to the request for further communication.
 ///   2. The request sending is started by the executor accessing the `HTTPExecutingRequest`'s
 ///     property `requestHead: HTTPRequestHead`. Based on the `requestHead` the executor can
 ///     determine if the request has a body (Is a "content-length" or "transfer-encoding"
@@ -201,18 +198,15 @@ protocol HTTPExecutingRequest: AnyObject {
 
     /// Will be called by the ChannelHandler to indicate that the request is going to be sent.
     ///
-    /// This will be called on the Channel's EventLoop. Do **not block** during your execution!
-    ///
-    /// - Returns: A bool indicating if the request should be started. Return `false` if the request
-    ///            has already been cancelled. If the request is cancelled after the `willExecuteRequest`
-    ///            method was called. The executing request must call `executor.cancel()` to stop
-    ///            request execution.
-    func willExecuteRequest(_: HTTPRequestExecutor) -> Bool
+    /// This will be called on the Channel's EventLoop. Do **not block** during your execution! If the
+    /// request is cancelled after the `willExecuteRequest` method was called. The executing
+    /// request must call `executor.cancel()` to stop request execution.
+    func willExecuteRequest(_: HTTPRequestExecutor)
 
     /// Will be called by the ChannelHandler to indicate that the request head has been sent.
     ///
     /// This will be called on the Channel's EventLoop. Do **not block** during your execution!
-    func requestHeadSent(_: HTTPRequestHead)
+    func requestHeadSent()
 
     /// Start or resume request body streaming
     ///
@@ -238,9 +232,9 @@ protocol HTTPExecutingRequest: AnyObject {
     /// be called in quick succession. It is the task's job to buffer those events for the user. Once all
     /// buffered data has been consumed the task must call `executor.demandResponseBodyStream`
     /// to ask for more data.
-    func receiveResponseBodyPart(_ buffer: ByteBuffer)
+    func receiveResponseBodyParts(_ buffer: CircularBuffer<ByteBuffer>)
 
-    func receiveResponseEnd()
+    func succeedRequest(_ buffer: CircularBuffer<ByteBuffer>?)
 
     func fail(_ error: Error)
 }

Diff for: Sources/AsyncHTTPClient/RequestBag+StateMachine.swift

@@ -43,7 +43,7 @@ extension RequestBag {
 
             case initialized
             case buffering(CircularBuffer<ByteBuffer>, next: Next)
-            case waitingForRemote(CircularBuffer<ByteBuffer>)
+            case waitingForRemote
         }
 
         private var state: State = .initialized
@@ -282,26 +282,31 @@ extension RequestBag.StateMachine {
         }
     }
 
-    mutating func receiveResponseBodyPart(_ byteBuffer: ByteBuffer) -> ByteBuffer? {
+    mutating func receiveResponseBodyParts(_ buffer: CircularBuffer<ByteBuffer>) -> ByteBuffer? {
         switch self.state {
         case .initialized, .queued:
             preconditionFailure("How can we receive a response body part, if the request hasn't started yet.")
         case .executing(_, _, .initialized):
             preconditionFailure("If we receive a response body, we must have received a head before")
 
-        case .executing(let executor, let requestState, .buffering(var buffer, next: let next)):
+        case .executing(let executor, let requestState, .buffering(var currentBuffer, next: let next)):
             guard case .askExecutorForMore = next else {
                 preconditionFailure("If we have received an error or eof before, why did we get another body part? Next: \(next)")
             }
 
             self.state = .modifying
-            buffer.append(byteBuffer)
-            self.state = .executing(executor, requestState, .buffering(buffer, next: next))
+            if currentBuffer.isEmpty {
+                currentBuffer = buffer
+            } else {
+                currentBuffer.append(contentsOf: buffer)
+            }
+            self.state = .executing(executor, requestState, .buffering(currentBuffer, next: next))
             return nil
-        case .executing(let executor, let requestState, .waitingForRemote(let buffer)):
-            assert(buffer.isEmpty, "If we wait for remote, the buffer must be empty")
+        case .executing(let executor, let requestState, .waitingForRemote):
+            var buffer = buffer
+            let first = buffer.removeFirst()
             self.state = .executing(executor, requestState, .buffering(buffer, next: .askExecutorForMore))
-            return byteBuffer
+            return first
         case .redirected:
             // ignore body
             return nil
@@ -315,30 +320,42 @@ extension RequestBag.StateMachine {
     }
 
     enum ReceiveResponseEndAction {
+        case consume(ByteBuffer)
         case redirect(RedirectHandler<Delegate.Response>, HTTPResponseHead, URL)
         case succeedRequest
         case none
     }
 
-    mutating func receiveResponseEnd() -> ReceiveResponseEndAction {
+    mutating func succeedRequest(_ newChunks: CircularBuffer<ByteBuffer>?) -> ReceiveResponseEndAction {
         switch self.state {
         case .initialized, .queued:
             preconditionFailure("How can we receive a response body part, if the request hasn't started yet.")
         case .executing(_, _, .initialized):
             preconditionFailure("If we receive a response body, we must have received a head before")
 
-        case .executing(let executor, let requestState, .buffering(let buffer, next: let next)):
+        case .executing(let executor, let requestState, .buffering(var buffer, next: let next)):
             guard case .askExecutorForMore = next else {
                 preconditionFailure("If we have received an error or eof before, why did we get another body part? Next: \(next)")
             }
 
+            if buffer.isEmpty, let newChunks = newChunks {
+                buffer = newChunks
+            } else if let newChunks = newChunks {
+                buffer.append(contentsOf: newChunks)
+            }
+
             self.state = .executing(executor, requestState, .buffering(buffer, next: .eof))
             return .none
 
-        case .executing(_, _, .waitingForRemote(let buffer)):
-            assert(buffer.isEmpty, "If we wait for remote, the buffer must be empty")
-            self.state = .finished(error: nil)
-            return .succeedRequest
+        case .executing(let executor, let requestState, .waitingForRemote):
+            guard var newChunks = newChunks, !newChunks.isEmpty else {
+                self.state = .finished(error: nil)
+                return .succeedRequest
+            }
+
+            let first = newChunks.removeFirst()
+            self.state = .executing(executor, requestState, .buffering(newChunks, next: .eof))
+            return .consume(first)
 
         case .redirected(let head, let redirectURL):
             self.state = .finished(error: nil)
@@ -421,7 +438,7 @@ extension RequestBag.StateMachine {
             }
 
             // buffer is empty, wait for more
-            self.state = .executing(executor, requestState, .waitingForRemote(buffer))
+            self.state = .executing(executor, requestState, .waitingForRemote)
             return .requestMoreFromExecutor(executor)
 
         case .executing(let executor, let requestState, .buffering(var buffer, next: .eof)):
@@ -482,7 +499,7 @@ extension RequestBag.StateMachine {
         case .executing(let executor, _, .initialized):
             self.state = .finished(error: error)
             return .failTask(nil, executor)
-        case .executing(let executor, _, .waitingForRemote(_)):
+        case .executing(let executor, _, .waitingForRemote):
             self.state = .finished(error: error)
             return .failTask(nil, executor)
         case .redirected: