Code review

Author: fabianfett

Sources/AsyncHTTPClient/ConnectionPool/HTTPExecutingRequest.swift

@@ -207,12 +207,12 @@ protocol HTTPExecutingRequest: AnyObject {
     ///            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
+    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 +238,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)
 }

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: