Add Separate Response Hooks (#3762)

### Issue Link 🔗
#3401

### Goals ⚽
This PR adds `onHTTPResponse` closure hooks to `DataRequest` /
`UploadRequest` and `DataStreamRequest` to enable the cancellation of
requests before data is transferred, requests that need to check
response info for later parsing, or peculiar requests that may trigger
multiple response callbacks, like MJPEG streams.

### Implementation Details 🚧
Like the other value hooks, this API accepts a single closure. The only
unique bit here is that there's second, disfavored, version of the API
that allows the user to return `ResponseDisposition` value to cancel or
end the request without the body.

### Testing Details 🔍
Streaming tests are being updated to check these events always fire
before the other events. More tests are needed around cancellation or
ending behavior to ensure that's really possible.

.swiftformat

@@ -10,7 +10,6 @@
 
 # format options
 
---closingparen same-line
 --commas inline
 --comments indent
 --decimalgrouping 3,5

Documentation/AdvancedUsage.md

@@ -435,9 +435,64 @@ AF.request(...)
     }
 ```
 
-#### Response
+#### `HTTPURLResponse`s
 
-Each `Request` may have an `HTTPURLResponse` value available once the request is complete. This value is only available if the request wasn’t cancelled and didn’t fail to make the network request. Additionally, if the request is retried, only the _last_ response is available. Intermediate responses can be derived from the `URLSessionTask`s in the `tasks` property.
+Alamofire receives `HTTPURLResponse` values from its underlying `URLSession` as the session starts the connection to the server. These values can be received in an `onHTTPResponse` closure and used to determine whether the request should continue or be cancelled. Both `DataRequest` / `UploadRequest` and `DataStreamRequest` provide these hooks. 
+
+In the case of `DataStreamRequest`, this event can be used in conjunction with the stream handlers to parse `HTTPURLResponse`s as part of the stream handling. To guarantee proper event ordering in that case, ensure the same `DispatchQueue` is used for both APIs. By default both APIs use `.main` for the completion handler, so this should only be a concern if you customize the queue used in either case.
+
+In the case of multiple `.onHTTPResponse` calls on a request, only the last one will be called.
+
+In it's simplest form `onHTTPResponse` simply provides the `HTTPURLResponse` value.
+
+```swift
+AF.request(...)
+    .onHTTPResponse { response in
+        print(response)
+    }
+```
+
+If control over the future of the request is desired, a completion handler value is provided which returns a `Request.ResponseDisposition` value.
+
+```swift
+AF.request(...)
+    .onHTTPResponse { response, completionHandler in
+        print(response)
+        completionHandler(.allow)
+    }
+```
+
+> The `completionHandler` MUST be called otherwise the request will hang until it times out.
+
+The `completionHandler` can also be used to cancel the request. This acts much like `cancel()` being called on the request itself.
+
+```swift
+AF.request(...)
+    .onHTTPResponse { response, completionHandler in
+        print(response)
+        completionHandler(.cancel)
+    }
+```
+
+Additionally, there are forms of both versions of `onHTTPResponse` available which provide `async` closures, allowing the use of `await` in the body. These versions are available on Swift 5.7 and above.
+
+```swift
+AF.request(...)
+    .onHTTPResponse { response in
+        await someAsyncMethod()
+        return .allow
+    }
+```
+
+Finally, an async stream of `HTTPURLResponse` values can also be used.
+
+```swift
+let responses = AF.request(...).httpResponses()
+
+for await response in responses {
+    print(response)
+}
+```
 
 #### `URLSessionTaskMetrics`
 

Source/Concurrency.swift

@@ -95,9 +95,9 @@ extension Request {
         }
     }
 
-    private func stream<T>(of type: T.Type = T.self,
-                           bufferingPolicy: StreamOf<T>.BufferingPolicy = .unbounded,
-                           yielder: @escaping (StreamOf<T>.Continuation) -> Void) -> StreamOf<T> {
+    fileprivate func stream<T>(of type: T.Type = T.self,
+                               bufferingPolicy: StreamOf<T>.BufferingPolicy = .unbounded,
+                               yielder: @escaping (StreamOf<T>.Continuation) -> Void) -> StreamOf<T> {
         StreamOf<T>(bufferingPolicy: bufferingPolicy) { [unowned self] continuation in
             yielder(continuation)
             // Must come after serializers run in order to catch retry progress.
@@ -168,6 +168,71 @@ public struct DataTask<Value> {
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
 extension DataRequest {
+    /// Creates a `StreamOf<HTTPURLResponse>` for the instance's responses.
+    ///
+    /// - Parameter bufferingPolicy: `BufferingPolicy` that determines the stream's buffering behavior.`.unbounded` by default.
+    ///
+    /// - Returns:                   The `StreamOf<HTTPURLResponse>`.
+    public func httpResponses(bufferingPolicy: StreamOf<HTTPURLResponse>.BufferingPolicy = .unbounded) -> StreamOf<HTTPURLResponse> {
+        stream(bufferingPolicy: bufferingPolicy) { [unowned self] continuation in
+            onHTTPResponse(on: underlyingQueue) { response in
+                continuation.yield(response)
+            }
+        }
+    }
+
+    #if swift(>=5.7)
+    /// Sets an async closure returning a `Request.ResponseDisposition`, called whenever the `DataRequest` produces an
+    /// `HTTPURLResponse`.
+    ///
+    /// - Note: Most requests will only produce a single response for each outgoing attempt (initial + retries).
+    ///         However, some types of response may trigger multiple `HTTPURLResponse`s, such as multipart streams,
+    ///         where responses after the first will contain the part headers.
+    ///
+    /// - Parameters:
+    ///   - handler: Async closure executed when a new `HTTPURLResponse` is received and returning a
+    ///              `ResponseDisposition` value. This value determines whether to continue the request or cancel it as
+    ///              if `cancel()` had been called on the instance. Note, this closure is called on an arbitrary thread,
+    ///              so any synchronous calls in it will execute in that context.
+    ///
+    /// - Returns:   The instance.
+    @_disfavoredOverload
+    @discardableResult
+    public func onHTTPResponse(
+        perform handler: @escaping @Sendable (_ response: HTTPURLResponse) async -> ResponseDisposition
+    ) -> Self {
+        onHTTPResponse(on: underlyingQueue) { response, completionHandler in
+            Task {
+                let disposition = await handler(response)
+                completionHandler(disposition)
+            }
+        }
+
+        return self
+    }
+
+    /// Sets an async closure called whenever the `DataRequest` produces an `HTTPURLResponse`.
+    ///
+    /// - Note: Most requests will only produce a single response for each outgoing attempt (initial + retries).
+    ///         However, some types of response may trigger multiple `HTTPURLResponse`s, such as multipart streams,
+    ///         where responses after the first will contain the part headers.
+    ///
+    /// - Parameters:
+    ///   - handler: Async closure executed when a new `HTTPURLResponse` is received. Note, this closure is called on an
+    ///              arbitrary thread, so any synchronous calls in it will execute in that context.
+    ///
+    /// - Returns:   The instance.
+    @discardableResult
+    public func onHTTPResponse(perform handler: @escaping @Sendable (_ response: HTTPURLResponse) async -> Void) -> Self {
+        onHTTPResponse { response in
+            await handler(response)
+            return .allow
+        }
+
+        return self
+    }
+    #endif
+
     /// Creates a `DataTask` to `await` a `Data` value.
     ///
     /// - Parameters:
@@ -625,6 +690,69 @@ public struct DataStreamTask {
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
 extension DataStreamRequest {
+    /// Creates a `StreamOf<HTTPURLResponse>` for the instance's responses.
+    ///
+    /// - Parameter bufferingPolicy: `BufferingPolicy` that determines the stream's buffering behavior.`.unbounded` by default.
+    ///
+    /// - Returns:                   The `StreamOf<HTTPURLResponse>`.
+    public func httpResponses(bufferingPolicy: StreamOf<HTTPURLResponse>.BufferingPolicy = .unbounded) -> StreamOf<HTTPURLResponse> {
+        stream(bufferingPolicy: bufferingPolicy) { [unowned self] continuation in
+            onHTTPResponse(on: underlyingQueue) { response in
+                continuation.yield(response)
+            }
+        }
+    }
+
+    #if swift(>=5.7)
+    /// Sets an async closure returning a `Request.ResponseDisposition`, called whenever the `DataStreamRequest`
+    /// produces an `HTTPURLResponse`.
+    ///
+    /// - Note: Most requests will only produce a single response for each outgoing attempt (initial + retries).
+    ///         However, some types of response may trigger multiple `HTTPURLResponse`s, such as multipart streams,
+    ///         where responses after the first will contain the part headers.
+    ///
+    /// - Parameters:
+    ///   - handler: Async closure executed when a new `HTTPURLResponse` is received and returning a
+    ///              `ResponseDisposition` value. This value determines whether to continue the request or cancel it as
+    ///              if `cancel()` had been called on the instance. Note, this closure is called on an arbitrary thread,
+    ///              so any synchronous calls in it will execute in that context.
+    ///
+    /// - Returns:   The instance.
+    @_disfavoredOverload
+    @discardableResult
+    public func onHTTPResponse(perform handler: @escaping @Sendable (HTTPURLResponse) async -> ResponseDisposition) -> Self {
+        onHTTPResponse(on: underlyingQueue) { response, completionHandler in
+            Task {
+                let disposition = await handler(response)
+                completionHandler(disposition)
+            }
+        }
+
+        return self
+    }
+
+    /// Sets an async closure called whenever the `DataStreamRequest` produces an `HTTPURLResponse`.
+    ///
+    /// - Note: Most requests will only produce a single response for each outgoing attempt (initial + retries).
+    ///         However, some types of response may trigger multiple `HTTPURLResponse`s, such as multipart streams,
+    ///         where responses after the first will contain the part headers.
+    ///
+    /// - Parameters:
+    ///   - handler: Async closure executed when a new `HTTPURLResponse` is received. Note, this closure is called on an
+    ///              arbitrary thread, so any synchronous calls in it will execute in that context.
+    ///
+    /// - Returns:   The instance.
+    @discardableResult
+    public func onHTTPResponse(perform handler: @escaping @Sendable (HTTPURLResponse) async -> Void) -> Self {
+        onHTTPResponse { response in
+            await handler(response)
+            return .allow
+        }
+
+        return self
+    }
+    #endif
+
     /// Creates a `DataStreamTask` used to `await` streams of serialized values.
     ///
     /// - Returns: The `DataStreamTask`.

Source/EventMonitor.swift

@@ -69,6 +69,9 @@ public protocol EventMonitor {
 
     // MARK: URLSessionDataDelegate Events
 
+    /// Event called during `URLSessionDataDelegate`'s `urlSession(_:dataTask:didReceive:completionHandler:)` method.
+    func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive response: URLResponse)
+
     /// Event called during `URLSessionDataDelegate`'s `urlSession(_:dataTask:didReceive:)` method.
     func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data)
 
@@ -244,6 +247,7 @@ extension EventMonitor {
                            didFinishCollecting metrics: URLSessionTaskMetrics) {}
     public func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {}
     public func urlSession(_ session: URLSession, taskIsWaitingForConnectivity task: URLSessionTask) {}
+    public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive response: URLResponse) {}
     public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {}
     public func urlSession(_ session: URLSession,
                            dataTask: URLSessionDataTask,
@@ -380,6 +384,10 @@ public final class CompositeEventMonitor: EventMonitor {
         performEvent { $0.urlSession(session, taskIsWaitingForConnectivity: task) }
     }
 
+    public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive response: URLResponse) {
+        performEvent { $0.urlSession(session, dataTask: dataTask, didReceive: response) }
+    }
+
     public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
         performEvent { $0.urlSession(session, dataTask: dataTask, didReceive: data) }
     }
@@ -593,6 +601,9 @@ open class ClosureEventMonitor: EventMonitor {
     /// Closure called on the `urlSession(_:taskIsWaitingForConnectivity:)` event.
     open var taskIsWaitingForConnectivity: ((URLSession, URLSessionTask) -> Void)?
 
+    /// Closure called on the `urlSession(_:dataTask:didReceive:completionHandler:)` event.
+    open var dataTaskDidReceiveResponse: ((URLSession, URLSessionDataTask, URLResponse) -> Void)?
+
     /// Closure that receives the `urlSession(_:dataTask:didReceive:)` event.
     open var dataTaskDidReceiveData: ((URLSession, URLSessionDataTask, Data) -> Void)?
 
@@ -741,6 +752,10 @@ open class ClosureEventMonitor: EventMonitor {
         taskIsWaitingForConnectivity?(session, task)
     }
 
+    open func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive response: URLResponse) {
+        dataTaskDidReceiveResponse?(session, dataTask, response)
+    }
+
     open func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
         dataTaskDidReceiveData?(session, dataTask, data)
     }

Source/NetworkReachabilityManager.swift

@@ -194,7 +194,8 @@ open class NetworkReachabilityManager {
                 let description = weakManager.manager?.flags?.readableDescription ?? "nil"
 
                 return Unmanaged.passRetained(description as CFString)
-            })
+            }
+        )
         let callback: SCNetworkReachabilityCallBack = { _, flags, info in
             guard let info = info else { return }