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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195 | import Foundation
import IrdP2PSignaling
@preconcurrency import WebRTC
/// The WebRTC-backed factory: Google WebRTC via the `WebRTC` xcframework.
public struct WebRTCPeerConnectionFactory: PeerConnectionFactory {
public init() {}
public func makePeerConnection(ice: IceConfiguration) async throws -> any PeerConnectionHandle {
try await makePeerConnection(ice: ice, continualGathering: true)
}
public func makePeerConnection(ice: IceConfiguration, continualGathering: Bool) async throws -> any PeerConnectionHandle {
guard let pc = WebRTCPeerConnection(ice: ice, continualGathering: continualGathering) else { throw P2PError.config("RTCPeerConnection could not be created") }
return pc
}
}
/// The only file that touches WebRTC types. Delegate callbacks arrive on WebRTC's
/// threads and are converted to `PeerConnectionEvent` values; every outbound call is
/// a continuation-wrapped WebRTC API call.
final class WebRTCPeerConnection: NSObject, PeerConnectionHandle, @unchecked Sendable, RTCPeerConnectionDelegate, RTCDataChannelDelegate {
private static let factory: RTCPeerConnectionFactory = {
RTCInitializeSSL()
return RTCPeerConnectionFactory()
}()
let events: AsyncStream<PeerConnectionEvent>
private let emit: AsyncStream<PeerConnectionEvent>.Continuation
private let pc: RTCPeerConnection
private let channel: RTCDataChannel
private let lock = NSLock()
private var closed = false
static func gatheringPolicy(continual: Bool) -> RTCContinualGatheringPolicy {
continual ? .gatherContinually : .gatherOnce
}
init?(ice: IceConfiguration, continualGathering: Bool = true) {
let config = RTCConfiguration()
config.sdpSemantics = .unifiedPlan
config.iceServers = ice.servers.map { RTCIceServer(urlStrings: $0.urls, username: $0.username, credential: $0.credential) }
config.iceTransportPolicy = ice.relayOnly ? .relay : .all
config.continualGatheringPolicy = WebRTCPeerConnection.gatheringPolicy(continual: continualGathering)
config.bundlePolicy = .maxBundle
config.rtcpMuxPolicy = .require
let constraints = RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil)
(events, emit) = AsyncStream<PeerConnectionEvent>.makeStream(bufferingPolicy: .unbounded)
guard let pc = WebRTCPeerConnection.factory.peerConnection(with: config, constraints: constraints, delegate: nil) else { return nil }
let dcConfig = RTCDataChannelConfiguration()
dcConfig.isNegotiated = true
dcConfig.channelId = 0
dcConfig.isOrdered = true
guard let channel = pc.dataChannel(forLabel: "ird", configuration: dcConfig) else { return nil }
self.pc = pc
self.channel = channel
super.init()
pc.delegate = self
channel.delegate = self
}
// MARK: PeerConnectionHandle
func createOffer(iceRestart: Bool) async throws -> String {
let constraints = RTCMediaConstraints(mandatoryConstraints: iceRestart ? [kRTCMediaConstraintsIceRestart: kRTCMediaConstraintsValueTrue] : nil, optionalConstraints: nil)
let sdp: RTCSessionDescription = try await withCheckedThrowingContinuation { c in
pc.offer(for: constraints) { sdp, error in
if let sdp { c.resume(returning: sdp) } else { c.resume(throwing: P2PError.transport("createOffer: \(error?.localizedDescription ?? "no sdp")")) }
}
}
try await setLocal(sdp)
return sdp.sdp
}
func createAnswer() async throws -> String {
let constraints = RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil)
let sdp: RTCSessionDescription = try await withCheckedThrowingContinuation { c in
pc.answer(for: constraints) { sdp, error in
if let sdp { c.resume(returning: sdp) } else { c.resume(throwing: P2PError.transport("createAnswer: \(error?.localizedDescription ?? "no sdp")")) }
}
}
try await setLocal(sdp)
return sdp.sdp
}
private func setLocal(_ sdp: RTCSessionDescription) async throws {
try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, any Error>) in
pc.setLocalDescription(sdp) { error in
if let error { c.resume(throwing: P2PError.transport("setLocalDescription: \(error.localizedDescription)")) } else { c.resume() }
}
}
}
func setRemoteDescription(_ type: SDPType, sdp: String) async throws {
let desc = RTCSessionDescription(type: type == .offer ? .offer : .answer, sdp: sdp)
try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, any Error>) in
pc.setRemoteDescription(desc) { error in
if let error { c.resume(throwing: P2PError.transport("setRemoteDescription: \(error.localizedDescription)")) } else { c.resume() }
}
}
}
func rollback() async throws {
try await setLocal(RTCSessionDescription(type: .rollback, sdp: ""))
}
func addIceCandidate(candidate: String, sdpMid: String?, sdpMLineIndex: Int?) async throws {
let c = RTCIceCandidate(sdp: candidate, sdpMLineIndex: Int32(sdpMLineIndex ?? 0), sdpMid: sdpMid)
try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, any Error>) in
pc.add(c) { error in
if let error { cont.resume(throwing: P2PError.transport("addIceCandidate: \(error.localizedDescription)")) } else { cont.resume() }
}
}
}
func signalingState() async -> SignalingState {
switch pc.signalingState {
case .stable: return .stable
case .haveLocalOffer: return .haveLocalOffer
case .haveRemoteOffer: return .haveRemoteOffer
case .closed: return .closed
default: return .other
}
}
func send(text: String) async throws {
guard channel.readyState == .open else { throw P2PError.transport("data channel not open") }
guard channel.sendData(RTCDataBuffer(data: Data(text.utf8), isBinary: false)) else { throw P2PError.transport("data channel send failed") }
}
func send(data: Data) async throws {
guard channel.readyState == .open else { throw P2PError.transport("data channel not open") }
guard channel.sendData(RTCDataBuffer(data: data, isBinary: true)) else { throw P2PError.transport("data channel send failed") }
}
func bufferedAmount() async -> Int { Int(channel.bufferedAmount) }
func close() async {
let already: Bool = lock.withLock { defer { closed = true }; return closed }
guard !already else { return }
channel.close()
pc.close()
emit.finish()
}
// MARK: RTCPeerConnectionDelegate
func peerConnection(_ peerConnection: RTCPeerConnection, didChange stateChanged: RTCSignalingState) {}
func peerConnection(_ peerConnection: RTCPeerConnection, didAdd stream: RTCMediaStream) {}
func peerConnection(_ peerConnection: RTCPeerConnection, didRemove stream: RTCMediaStream) {}
func peerConnectionShouldNegotiate(_ peerConnection: RTCPeerConnection) {}
func peerConnection(_ peerConnection: RTCPeerConnection, didChange newState: RTCIceConnectionState) {}
func peerConnection(_ peerConnection: RTCPeerConnection, didRemove candidates: [RTCIceCandidate]) {}
func peerConnection(_ peerConnection: RTCPeerConnection, didOpen dataChannel: RTCDataChannel) {}
func peerConnection(_ peerConnection: RTCPeerConnection, didChange newState: RTCIceGatheringState) {
if newState == .complete { emit.yield(.iceGatheringComplete) }
}
func peerConnection(_ peerConnection: RTCPeerConnection, didGenerate candidate: RTCIceCandidate) {
emit.yield(.iceCandidate(candidate: candidate.sdp, sdpMid: candidate.sdpMid, sdpMLineIndex: Int(candidate.sdpMLineIndex)))
}
func peerConnection(_ peerConnection: RTCPeerConnection, didChange newState: RTCPeerConnectionState) {
let mapped: PeerConnectionState
switch newState {
case .new: mapped = .new
case .connecting: mapped = .connecting
case .connected: mapped = .connected
case .disconnected: mapped = .disconnected
case .failed: mapped = .failed
case .closed: mapped = .closed
@unknown default: mapped = .new
}
emit.yield(.connectionState(mapped))
}
// MARK: RTCDataChannelDelegate
func dataChannelDidChangeState(_ dataChannel: RTCDataChannel) {
switch dataChannel.readyState {
case .open: emit.yield(.dataChannelOpen)
case .closed: emit.yield(.dataChannelClosed)
default: break
}
}
func dataChannel(_ dataChannel: RTCDataChannel, didReceiveMessageWith buffer: RTCDataBuffer) {
emit.yield(.message(buffer.data, isBinary: buffer.isBinary))
}
func dataChannel(_ dataChannel: RTCDataChannel, didChangeBufferedAmount amount: UInt64) {
if amount < 256 * 1024 { emit.yield(.bufferedAmountLow) }
}
}
|