Apply suggestions from code review

Co-authored-by: George Barnett <[email protected]>

Author: fabianfett

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

@@ -149,7 +149,7 @@ protocol HTTPScheduledRequest: AnyObject {
     /// A connection to run this task on needs to be found before this deadline!
     var connectionDeadline: NIODeadline { get }
 
-    /// The task's eventLoopPreference
+    /// The task's `EventLoop` preference
     var eventLoopPreference: HTTPClient.EventLoopPreference { get }
 
     /// Informs the task, that it was queued for execution
@@ -196,37 +196,34 @@ protocol HTTPExecutingRequest: AnyObject {
     /// after `requestHeadSent` was called.
     var requestHead: HTTPRequestHead { get }
 
-    /// The maximal `TimeAmount` that is allowed to pass between reads from the Channel.
+    /// The maximal `TimeAmount` that is allowed to pass between `channelRead`s from the Channel.
     var idleReadTimeout: TimeAmount? { get }
 
-    /// Will be called by the ChannelHandler to indicate that the request is going to be send.
+    /// 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 really be started. Return false if the request has already been cancelled.
-    ///            If the request is cancelled after this method call `executor.cancel()` to stop request 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
 
     /// 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)
 
-    /// Start request streaming
+    /// Start or resume request body streaming
     ///
     /// This will be called on the Channel's EventLoop. Do **not block** during your execution!
-    func startRequestBodyStream()
+    func resumeRequestBodyStream()
 
     /// Pause request streaming
     ///
     /// This will be called on the Channel's EventLoop. Do **not block** during your execution!
     func pauseRequestBodyStream()
 
-    /// Resume request streaming
-    ///
-    /// This will be called on the Channel's EventLoop. Do **not block** during your execution!
-    func resumeRequestBodyStream()
-
     /// Receive a response head.
     ///
     /// Please note that `receiveResponseHead` and `receiveResponseBodyPart` may

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

@@ -92,39 +92,45 @@ extension RequestBag.StateMachine {
         }
     }
 
-    enum StartProducingAction {
-        case startWriter(HTTPClient.Body.StreamWriter, body: HTTPClient.Body)
+    enum ResumeProducingAction {
+        case startWriter
+        case succeedBackpressurePromise(EventLoopPromise<Void>?)
         case none
     }
 
-    mutating func startRequestBodyStream(
-        _ body: HTTPClient.Body?,
-        writeMethod: @escaping (IOData) -> EventLoopFuture<Void>
-    ) -> StartProducingAction {
+    mutating func resumeRequestBodyStream() -> ResumeProducingAction {
         switch self.state {
+        case .initialized, .queued:
+            preconditionFailure("A request stream can only be resumed, if the request was started")
+
         case .executing(let executor, .initialized, .initialized):
-            guard let body = body else {
-                preconditionFailure("Did not expect to get a call to `startRequestBodyStream`, if there is no body.")
-            }
+            self.state = .executing(executor, .producing, .initialized)
+            return .startWriter
 
-            let streamWriter = HTTPClient.Body.StreamWriter { part -> EventLoopFuture<Void> in
-                writeMethod(part)
-            }
+        case .executing(_, .producing, _):
+            preconditionFailure("Expected that resume is only called when if we were paused before")
 
-            self.state = .executing(executor, .producing, .initialized)
+        case .executing(let executor, .paused(let promise), let responseState):
+            self.state = .executing(executor, .producing, responseState)
+            return .succeedBackpressurePromise(promise)
 
-            return .startWriter(streamWriter, body: body)
+        case .executing(_, .finished, _):
+            // the channels writability changed to writable after we have forwarded all the
+            // request bytes. Can be ignored.
+            return .none
 
-        case .finished(error: .some):
+        case .executing(_, .initialized, .buffering), .executing(_, .initialized, .waitingForRemote):
+            preconditionFailure("Invalid states: Response can not be received before request")
+
+        case .redirected:
+            // if we are redirected, we should cancel our request body stream anyway
             return .none
 
-        case .initialized,
-             .queued,
-             .executing(_, _, _),
-             .finished(error: .none),
-             .redirected,
-             .modifying:
-            preconditionFailure("Expected the state to be either initialized or failed. Invalid state: \(self.state)")
+        case .finished:
+            preconditionFailure("Invalid state")
+
+        case .modifying:
+            preconditionFailure("Invalid state")
         }
     }
 
@@ -156,37 +162,6 @@ extension RequestBag.StateMachine {
         }
     }
 
-    mutating func resumeRequestBodyStream() -> EventLoopPromise<Void>? {
-        switch self.state {
-        case .initialized, .queued:
-            preconditionFailure("A request stream can only be resumed, if the request was started")
-        case .executing(let executor, let requestState, let responseState):
-            switch requestState {
-            case .initialized:
-                preconditionFailure("Request stream must be started before it can be paused")
-            case .producing:
-                preconditionFailure("Expected that resume is only called when if we were paused before")
-            case .paused(let promise):
-                self.state = .executing(executor, .producing, responseState)
-                return promise
-            case .finished:
-                // the channels writability changed to writable after we have forwarded all the
-                // request bytes. Can be ignored.
-                return nil
-            }
-
-        case .redirected:
-            // if we are redirected, we should cancel our request body stream anyway
-            return nil
-
-        case .finished:
-            preconditionFailure("Invalid state")
-
-        case .modifying:
-            preconditionFailure("Invalid state")
-        }
-    }
-
     enum WriteAction {
         case write(IOData, HTTPRequestExecutor, EventLoopFuture<Void>)
 
@@ -195,10 +170,6 @@ extension RequestBag.StateMachine {
     }
 
     mutating func writeNextRequestPart(_ part: IOData, taskEventLoop: EventLoop) -> WriteAction {
-        // this method is invoked with bodyPart and returns a future that signals that
-        // more data can be send.
-        // it may be invoked on any eventLoop
-
         switch self.state {
         case .initialized, .queued:
             preconditionFailure("Invalid state: \(self.state)")
@@ -210,6 +181,8 @@ extension RequestBag.StateMachine {
                 return .write(part, executor, taskEventLoop.makeSucceededFuture(()))
 
             case .paused(.none):
+                // backpressure is signaled to the writer using unfulfilled futures. if there
+                // is no existing, unfulfilled promise, let's create a new one
                 let promise = taskEventLoop.makePromise(of: Void.self)
                 self.state = .executing(executor, .paused(promise), responseState)
                 return .write(part, executor, promise.futureResult)
@@ -459,7 +432,6 @@ extension RequestBag.StateMachine {
                 return .consume(byteBuffer)
             }
 
-            // buffer is empty, wait for more
             self.state = .finished(error: nil)
             return .finishStream
 

Diff for: Sources/AsyncHTTPClient/RequestBag.swift

@@ -115,32 +115,38 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>: HTTPScheduledReque
         }
     }
 
-    enum StartProducingAction {
-        case startWriter(HTTPClient.Body.StreamWriter, body: HTTPClient.Body)
-        case none
-    }
-
-    func startRequestBodyStream() {
+    func resumeRequestBodyStream() {
         let produceAction = self.stateLock.withLock {
-            self._state.startRequestBodyStream(self.request.body, writeMethod: self.writeNextRequestPart)
+            self._state.resumeRequestBodyStream()
         }
 
         switch produceAction {
-        case .startWriter(let writer, body: let body):
-            func startWriter0(_ writer: HTTPClient.Body.StreamWriter, body: HTTPClient.Body) {
+        case .startWriter:
+            func startWriter0() {
+                guard let body = self.request.body else {
+                    preconditionFailure("Expected to have a body, if the `HTTPRequestStateMachine` resume a request stream")
+                }
+
+                let writer = HTTPClient.Body.StreamWriter {
+                    self.writeNextRequestPart($0)
+                }
+
                 body.stream(writer).whenComplete {
                     self.finishRequestBodyStream($0)
                 }
             }
 
             if self.task.eventLoop.inEventLoop {
-                startWriter0(writer, body: body)
+                startWriter0()
             } else {
                 return self.task.eventLoop.execute {
-                    startWriter0(writer, body: body)
+                    startWriter0()
                 }
             }
 
+        case .succeedBackpressurePromise(let promise):
+            promise?.succeed(())
+
         case .none:
             break
         }
@@ -152,14 +158,6 @@ final class RequestBag<Delegate: HTTPClientResponseDelegate>: HTTPScheduledReque
         }
     }
 
-    func resumeRequestBodyStream() {
-        let promise = self.stateLock.withLock {
-            self._state.resumeRequestBodyStream()
-        }
-
-        promise?.succeed(())
-    }
-
     enum WriteAction {
         case write(IOData, HTTPRequestExecutor, EventLoopFuture<Void>)
 

Diff for: Tests/AsyncHTTPClientTests/RequestBagTests.swift

@@ -74,7 +74,7 @@ final class RequestBagTests: XCTestCase {
         bag.requestHeadSent(bag.requestHead)
         XCTAssertEqual(delegate.hitDidSendRequestHead, 1)
         streamIsAllowedToWrite = true
-        bag.startRequestBodyStream()
+        bag.resumeRequestBodyStream()
         streamIsAllowedToWrite = false
 
         // after starting the body stream we should have received two writes