1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106 | import Foundation
/// `URLSessionWebSocketTask`-backed transport. The upgrade is awaited through the
/// session delegate so a 403/404/429 surfaces as `upgradeRejected` before any frame
/// is sent; protocol-level pings from the server are answered by URLSession itself.
public struct URLSessionWebSocketTransport: WebSocketTransport {
let configuration: URLSessionConfiguration
public init(configuration: URLSessionConfiguration = .default) {
self.configuration = configuration
}
public func connect(_ request: URLRequest) async throws -> any WebSocketConnection {
let delegate = OpenDelegate()
let session = URLSession(configuration: configuration, delegate: delegate, delegateQueue: nil)
let task = session.webSocketTask(with: request)
do {
try await delegate.waitForOpen(task: task)
} catch {
session.finishTasksAndInvalidate()
throw error
}
return Connection(task: task, session: session)
}
final class OpenDelegate: NSObject, URLSessionWebSocketDelegate, URLSessionTaskDelegate, @unchecked Sendable {
private let lock = NSLock()
private var continuation: CheckedContinuation<Void, any Error>?
private var finished = false
func waitForOpen(task: URLSessionWebSocketTask) async throws {
try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, any Error>) in
lock.lock()
continuation = c
lock.unlock()
task.resume()
}
}
private func finish(_ result: Result<Void, any Error>) {
lock.lock()
let c = continuation
continuation = nil
let already = finished
finished = true
lock.unlock()
guard !already, let c else { return }
c.resume(with: result)
}
func urlSession(_ session: URLSession, webSocketTask: URLSessionWebSocketTask, didOpenWithProtocol protocol: String?) {
finish(.success(()))
}
func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: (any Error)?) {
if let http = task.response as? HTTPURLResponse, http.statusCode != 101 {
finish(.failure(P2PError.upgradeRejected(status: http.statusCode)))
} else if let error {
finish(.failure(URLSessionWebSocketTransport.map(error)))
} else {
finish(.failure(P2PError.closed))
}
}
}
static func map(_ error: any Error) -> P2PError {
if let e = error as? P2PError { return e }
if let e = error as? URLError {
if e.code == .timedOut { return .timeout }
if e.code == .cancelled { return .closed }
}
return .transport(error.localizedDescription)
}
final class Connection: WebSocketConnection, @unchecked Sendable {
let task: URLSessionWebSocketTask
let session: URLSession
init(task: URLSessionWebSocketTask, session: URLSession) {
self.task = task
self.session = session
}
func send(text: String) async throws {
do { try await task.send(.string(text)) } catch { throw URLSessionWebSocketTransport.map(error) }
}
func receive() async throws -> WebSocketMessage {
do {
switch try await task.receive() {
case .string(let s): return .text(s)
case .data(let d): return .data(d)
@unknown default: throw P2PError.transport("unknown WebSocket message type")
}
} catch {
if task.closeCode != .invalid { throw P2PError.closed }
throw URLSessionWebSocketTransport.map(error)
}
}
func close(code: Int, reason: String?) async {
task.cancel(with: URLSessionWebSocketTask.CloseCode(rawValue: code) ?? .normalClosure, reason: reason.map { Data($0.utf8) })
session.finishTasksAndInvalidate()
}
}
}
|