21674187555

Committed 04 Feb 2026 01:55PM UTC coverage: 80.045% (-0.3%) from 80.34%

Build # 21674187555

Build Type

Pull #903

github

Committed by

web-flow

Commit Message

Merge 9c93b1196 into 63eec095d

Pull Request Pull Request #903: feat(functions): add proper streaming response support to HTTPClient

Coverage Stats

144 of 198 new or added lines in 4 files covered. (72.73%)

2 existing lines in 1 file now uncovered.

6430 of 8033 relevant lines covered (80.04%)

27.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.58

/Sources/Helpers/HTTP/HTTPClient.swift

//
//  HTTPClient.swift
//
//
//  Created by Guilherme Souza on 30/04/24.
//

import Foundation
import HTTPTypes

#if canImport(FoundationNetworking)
  import FoundationNetworking
#endif

package protocol HTTPClientType: Sendable {
  func send(_ request: HTTPRequest) async throws -> HTTPResponse
  func sendStreaming(_ request: HTTPRequest) async throws -> HTTPResponse.Stream
}

package actor HTTPClient: HTTPClientType {
  let fetch: @Sendable (URLRequest) async throws -> (Data, URLResponse)
  let interceptors: [any HTTPClientInterceptor]
  let sessionConfiguration: URLSessionConfiguration

  package init(
    fetch: @escaping @Sendable (URLRequest) async throws -> (Data, URLResponse),
    interceptors: [any HTTPClientInterceptor],
    sessionConfiguration: URLSessionConfiguration = .default
  ) {
    self.fetch = fetch
    self.interceptors = interceptors
    self.sessionConfiguration = sessionConfiguration
  }

  package func send(_ request: HTTPRequest) async throws -> HTTPResponse {
    var next: @Sendable (HTTPRequest) async throws -> HTTPResponse = { _request in
      let urlRequest = _request.urlRequest
      let (data, response) = try await self.fetch(urlRequest)
      guard let httpURLResponse = response as? HTTPURLResponse else {
        throw URLError(.badServerResponse)
      }
      return HTTPResponse(data: data, response: httpURLResponse)
    }

    for interceptor in interceptors.reversed() {
      let tmp = next
      next = {
        try await interceptor.intercept($0, next: tmp)
      }
    }

    return try await next(request)
  }

  package func sendStreaming(_ request: HTTPRequest) async throws -> HTTPResponse.Stream {
    // Apply request-phase interceptors (modify headers, log request start, etc.)
    var modifiedRequest = request
    for interceptor in interceptors {
      modifiedRequest = try await interceptor.interceptRequest(modifiedRequest)
    }

    let urlRequest = modifiedRequest.urlRequest
    let capturedInterceptors = interceptors
    let capturedSessionConfiguration = sessionConfiguration

    return try await withCheckedThrowingContinuation {
      (continuation: CheckedContinuation<HTTPResponse.Stream, any Error>) in
      let delegate = StreamingResponseDelegate(
        request: modifiedRequest,
        interceptors: capturedInterceptors,
        continuation: continuation
      )

      let session = URLSession(
        configuration: capturedSessionConfiguration,
        delegate: delegate,
        delegateQueue: nil
      )

      let task = session.dataTask(with: urlRequest)
      delegate.setTask(task)
      task.resume()
    }
  }
}

package protocol HTTPClientInterceptor: Sendable {
  func intercept(
    _ request: HTTPRequest,
    next: @Sendable (HTTPRequest) async throws -> HTTPResponse
  ) async throws -> HTTPResponse

  /// Intercept only the request phase (before sending).
  /// Used for streaming requests where the response cannot be intercepted in the same way.
  /// Default implementation returns the request unmodified.
  func interceptRequest(_ request: HTTPRequest) async throws -> HTTPRequest

  /// Called when a streaming response completes (successfully or with error).
  /// Used for logging the completion of streaming requests.
  func onStreamingResponseComplete(_ request: HTTPRequest, error: (any Error)?) async
}

extension HTTPClientInterceptor {
  package func interceptRequest(_ request: HTTPRequest) async throws -> HTTPRequest {
    request
  }

  package func onStreamingResponseComplete(_ request: HTTPRequest, error: (any Error)?) async {
    // Default: no-op
  }
}

/// URLSession delegate for handling streaming responses.
final class StreamingResponseDelegate: NSObject, URLSessionDataDelegate, @unchecked Sendable {
  private let request: HTTPRequest
  private let interceptors: [any HTTPClientInterceptor]
  private let responseContinuation: CheckedContinuation<HTTPResponse.Stream, any Error>
  private let lock = NSLock()

  private var task: URLSessionTask?
  private var dataContinuation: AsyncThrowingStream<Data, any Error>.Continuation?
  private var hasResumedResponseContinuation = false

  init(
    request: HTTPRequest,
    interceptors: [any HTTPClientInterceptor],
    continuation: CheckedContinuation<HTTPResponse.Stream, any Error>
  ) {
    self.request = request
    self.interceptors = interceptors
    self.responseContinuation = continuation
    super.init()
  }

  func setTask(_ task: URLSessionTask) {
    lock.lock()
    self.task = task
    lock.unlock()
  }

  func urlSession(
    _ session: URLSession,
    dataTask: URLSessionDataTask,
    didReceive response: URLResponse,
    completionHandler: @escaping (URLSession.ResponseDisposition) -> Void
  ) {
    guard let httpResponse = response as? HTTPURLResponse else {
      lock.lock()
      if !hasResumedResponseContinuation {
        hasResumedResponseContinuation = true
        lock.unlock()
        responseContinuation.resume(throwing: URLError(.badServerResponse))
      } else {
        lock.unlock()
      }
      completionHandler(.cancel)
      return
    }

    let (stream, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()

    lock.lock()
    dataContinuation = continuation
    hasResumedResponseContinuation = true
    lock.unlock()

    let capturedRequest = request
    let capturedInterceptors = interceptors

    continuation.onTermination = { [weak self] _ in
      guard let self else { return }
      self.lock.lock()
      let task = self.task
      self.lock.unlock()
      task?.cancel()

      // Notify interceptors of completion
      Task {
        for interceptor in capturedInterceptors {
          await interceptor.onStreamingResponseComplete(capturedRequest, error: nil)
        }
      }
    }

    let streamResponse = HTTPResponse.Stream(
      statusCode: httpResponse.statusCode,
      headers: HTTPFields(httpResponse.allHeaderFields as? [String: String] ?? [:]),
      underlyingResponse: httpResponse,
      body: stream
    )

    responseContinuation.resume(returning: streamResponse)
    completionHandler(.allow)
  }

  func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
    lock.lock()
    let continuation = dataContinuation
    lock.unlock()
    continuation?.yield(data)
  }

  func urlSession(
    _ session: URLSession, task: URLSessionTask, didCompleteWithError error: (any Error)?
  ) {
    lock.lock()
    let continuation = dataContinuation
    let alreadyResumed = hasResumedResponseContinuation
    lock.unlock()

    if let error {
      if !alreadyResumed {
        lock.lock()
        hasResumedResponseContinuation = true
        lock.unlock()
        responseContinuation.resume(throwing: error)
      } else {
        continuation?.finish(throwing: error)
      }
    } else {
      continuation?.finish()
    }

    // Notify interceptors of completion
    let capturedRequest = request
    Task {
      for interceptor in interceptors {
        await interceptor.onStreamingResponseComplete(capturedRequest, error: error)
      }
    }
  }
}

1	//
2	// HTTPClient.swift
3	//
4	//
5	// Created by Guilherme Souza on 30/04/24.
6	//
7
8	import Foundation
9	import HTTPTypes
10
11	#if canImport(FoundationNetworking)
12	import FoundationNetworking
13	#endif
14
15	package protocol HTTPClientType: Sendable {
16	func send(_ request: HTTPRequest) async throws -> HTTPResponse
17	func sendStreaming(_ request: HTTPRequest) async throws -> HTTPResponse.Stream
18	}
19
20	package actor HTTPClient: HTTPClientType {
21	let fetch: @Sendable (URLRequest) async throws -> (Data, URLResponse)
22	let interceptors: [any HTTPClientInterceptor]
23	let sessionConfiguration: URLSessionConfiguration
24
25	package init(
26	fetch: @escaping @Sendable (URLRequest) async throws -> (Data, URLResponse),
27	interceptors: [any HTTPClientInterceptor],
28	sessionConfiguration: URLSessionConfiguration = .default
29	) {	418✔
30	self.fetch = fetch	418✔
31	self.interceptors = interceptors	418✔
32	self.sessionConfiguration = sessionConfiguration	418✔
33	}	418✔
34
35	package func send(_ request: HTTPRequest) async throws -> HTTPResponse {	249✔
36	var next: @Sendable (HTTPRequest) async throws -> HTTPResponse = { _request in	249✔
37	let urlRequest = _request.urlRequest	249✔
38	let (data, response) = try await self.fetch(urlRequest)	249✔
39	guard let httpURLResponse = response as? HTTPURLResponse else {	209✔
40	throw URLError(.badServerResponse)	32✔
41	}	177✔
42	return HTTPResponse(data: data, response: httpURLResponse)	177✔
43	}	249✔
44		249✔
45	for interceptor in interceptors.reversed() {	249✔
46	let tmp = next	107✔
47	next = {	107✔
48	try await interceptor.intercept($0, next: tmp)	107✔
49	}	73✔
50	}	249✔
51		249✔
52	return try await next(request)	249✔
53	}	249✔
54
55	package func sendStreaming(_ request: HTTPRequest) async throws -> HTTPResponse.Stream {	3✔
56	// Apply request-phase interceptors (modify headers, log request start, etc.)	3✔
57	var modifiedRequest = request	3✔
58	for interceptor in interceptors {	3✔
NEW 59	modifiedRequest = try await interceptor.interceptRequest(modifiedRequest)	×
60	}	3✔
61		3✔
62	let urlRequest = modifiedRequest.urlRequest	3✔
63	let capturedInterceptors = interceptors	3✔
64	let capturedSessionConfiguration = sessionConfiguration	3✔
65		3✔
66	return try await withCheckedThrowingContinuation {	3✔
67	(continuation: CheckedContinuation<HTTPResponse.Stream, any Error>) in	3✔
68	let delegate = StreamingResponseDelegate(	3✔
69	request: modifiedRequest,	3✔
70	interceptors: capturedInterceptors,	3✔
71	continuation: continuation	3✔
72	)	3✔
73		3✔
74	let session = URLSession(	3✔
75	configuration: capturedSessionConfiguration,	3✔
76	delegate: delegate,	3✔
77	delegateQueue: nil	3✔
78	)	3✔
79		3✔
80	let task = session.dataTask(with: urlRequest)	3✔
81	delegate.setTask(task)	3✔
82	task.resume()	3✔
83	}	3✔
84	}	3✔
85	}
86
87	package protocol HTTPClientInterceptor: Sendable {
88	func intercept(
89	_ request: HTTPRequest,
90	next: @Sendable (HTTPRequest) async throws -> HTTPResponse
91	) async throws -> HTTPResponse
92
93	/// Intercept only the request phase (before sending).
94	/// Used for streaming requests where the response cannot be intercepted in the same way.
95	/// Default implementation returns the request unmodified.
96	func interceptRequest(_ request: HTTPRequest) async throws -> HTTPRequest
97
98	/// Called when a streaming response completes (successfully or with error).
99	/// Used for logging the completion of streaming requests.
100	func onStreamingResponseComplete(_ request: HTTPRequest, error: (any Error)?) async
101	}
102
103	extension HTTPClientInterceptor {
NEW 104	package func interceptRequest(_ request: HTTPRequest) async throws -> HTTPRequest {	×
NEW 105	request	×
NEW 106	}	×
107
NEW 108	package func onStreamingResponseComplete(_ request: HTTPRequest, error: (any Error)?) async {	×
NEW 109	// Default: no-op	×
NEW 110	}	×
111	}
112
113	/// URLSession delegate for handling streaming responses.
114	final class StreamingResponseDelegate: NSObject, URLSessionDataDelegate, @unchecked Sendable {
115	private let request: HTTPRequest
116	private let interceptors: [any HTTPClientInterceptor]
117	private let responseContinuation: CheckedContinuation<HTTPResponse.Stream, any Error>
118	private let lock = NSLock()	3✔
119
120	private var task: URLSessionTask?
121	private var dataContinuation: AsyncThrowingStream<Data, any Error>.Continuation?
122	private var hasResumedResponseContinuation = false	3✔
123
124	init(
125	request: HTTPRequest,
126	interceptors: [any HTTPClientInterceptor],
127	continuation: CheckedContinuation<HTTPResponse.Stream, any Error>
128	) {	3✔
129	self.request = request	3✔
130	self.interceptors = interceptors	3✔
131	self.responseContinuation = continuation	3✔
132	super.init()	3✔
133	}	3✔
134
135	func setTask(_ task: URLSessionTask) {	3✔
136	lock.lock()	3✔
137	self.task = task	3✔
138	lock.unlock()	3✔
139	}	3✔
140
141	func urlSession(
142	_ session: URLSession,
143	dataTask: URLSessionDataTask,
144	didReceive response: URLResponse,
145	completionHandler: @escaping (URLSession.ResponseDisposition) -> Void
146	) {	3✔
147	guard let httpResponse = response as? HTTPURLResponse else {	3✔
NEW 148	lock.lock()	×
NEW 149	if !hasResumedResponseContinuation {	×
NEW 150	hasResumedResponseContinuation = true	×
NEW 151	lock.unlock()	×
NEW 152	responseContinuation.resume(throwing: URLError(.badServerResponse))	×
NEW 153	} else {	×
NEW 154	lock.unlock()	×
NEW 155	}	×
NEW 156	completionHandler(.cancel)	×
NEW 157	return	×
158	}	3✔
159		3✔
160	let (stream, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()	3✔
161		3✔
162	lock.lock()	3✔
163	dataContinuation = continuation	3✔
164	hasResumedResponseContinuation = true	3✔
165	lock.unlock()	3✔
166		3✔
167	let capturedRequest = request	3✔
168	let capturedInterceptors = interceptors	3✔
169		3✔
170	continuation.onTermination = { [weak self] _ in	3✔
171	guard let self else { return }	3✔
172	self.lock.lock()	3✔
173	let task = self.task	3✔
174	self.lock.unlock()	3✔
175	task?.cancel()	3✔
176		3✔
177	// Notify interceptors of completion	3✔
178	Task {	3✔
179	for interceptor in capturedInterceptors {	3✔
NEW 180	await interceptor.onStreamingResponseComplete(capturedRequest, error: nil)	×
181	}	3✔
182	}	3✔
183	}	3✔
184		3✔
185	let streamResponse = HTTPResponse.Stream(	3✔
186	statusCode: httpResponse.statusCode,	3✔
187	headers: HTTPFields(httpResponse.allHeaderFields as? [String: String] ?? [:]),	3✔
188	underlyingResponse: httpResponse,	3✔
189	body: stream	3✔
190	)	3✔
191		3✔
192	responseContinuation.resume(returning: streamResponse)	3✔
193	completionHandler(.allow)	3✔
194	}	3✔
195
196	func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {	1✔
197	lock.lock()	1✔
198	let continuation = dataContinuation	1✔
199	lock.unlock()	1✔
200	continuation?.yield(data)	1✔
201	}	1✔
202
203	func urlSession(
204	_ session: URLSession, task: URLSessionTask, didCompleteWithError error: (any Error)?
205	) {	3✔
206	lock.lock()	3✔
207	let continuation = dataContinuation	3✔
208	let alreadyResumed = hasResumedResponseContinuation	3✔
209	lock.unlock()	3✔
210		3✔
211	if let error {	3✔
NEW 212	if !alreadyResumed {	×
NEW 213	lock.lock()	×
NEW 214	hasResumedResponseContinuation = true	×
NEW 215	lock.unlock()	×
NEW 216	responseContinuation.resume(throwing: error)	×
NEW 217	} else {	×
NEW 218	continuation?.finish(throwing: error)	×
NEW 219	}	×
220	} else {	3✔
221	continuation?.finish()	3✔
222	}	3✔
223		3✔
224	// Notify interceptors of completion	3✔
225	let capturedRequest = request	3✔
226	Task {	3✔
227	for interceptor in interceptors {	3✔
NEW 228	await interceptor.onStreamingResponseComplete(capturedRequest, error: error)	×
229	}	3✔
230	}	3✔
231	}	3✔
232	}

supabase / supabase-swift / 21674187555

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous