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 | import Foundation
import IrdP2PSignaling
/// Pairs peer connections inside one process. Offers and answers carry the
/// handle's id, so two handles that exchange them through the room's signaling
/// end up connected exactly like real WebRTC peers would — with no network.
public final class InMemoryPeerConnectionFactory: PeerConnectionFactory, Sendable {
let network = InMemoryNetwork()
public init() {}
public func makePeerConnection(ice: IceConfiguration) async throws -> any PeerConnectionHandle {
let h = InMemoryPeerConnection(network: network)
await network.register(h)
return h
}
/// Breaks every established pair as if the network path died.
public func failAll() async { await network.failAll() }
/// Makes the next `count` connection attempts fail after negotiation instead of connecting.
public func failNextConnections(_ count: Int) async { await network.failNext(count) }
/// How many negotiations reached the connect step (successful or failed).
public var connectionAttempts: Int { get async { await network.attempts } }
}
actor InMemoryNetwork {
private var handles: [String: InMemoryPeerConnection] = [:]
private var pairs: [String: String] = [:]
private var failuresLeft = 0
private(set) var attempts = 0
func register(_ h: InMemoryPeerConnection) { handles[h.id] = h }
func unregister(_ id: String) {
handles[id] = nil
if let other = pairs.removeValue(forKey: id) {
pairs[other] = nil
handles[other]?.peerClosed()
}
}
func failNext(_ n: Int) { failuresLeft += n }
func failAll() {
let all = pairs
pairs.removeAll()
for (a, _) in all { handles[a]?.fail() }
}
/// Called by the answerer once it has both descriptions: connects both ends.
func connect(_ a: String, _ b: String) {
guard let ha = handles[a], let hb = handles[b] else { return }
attempts += 1
if failuresLeft > 0 {
failuresLeft -= 1
ha.fail(); hb.fail()
return
}
pairs[a] = b
pairs[b] = a
ha.connected(); hb.connected()
}
func deliver(from: String, _ data: Data, isBinary: Bool) {
guard let to = pairs[from], let h = handles[to] else { return }
h.receive(data, isBinary: isBinary)
}
func isConnected(_ id: String) -> Bool { pairs[id] != nil }
}
final class InMemoryPeerConnection: PeerConnectionHandle, @unchecked Sendable {
let id = UUID().uuidString
let events: AsyncStream<PeerConnectionEvent>
private let emit: AsyncStream<PeerConnectionEvent>.Continuation
private let network: InMemoryNetwork
private let lock = NSLock()
private var state: SignalingState = .stable
private var remoteId: String?
private var open = false
private var closed = false
init(network: InMemoryNetwork) {
self.network = network
(events, emit) = AsyncStream<PeerConnectionEvent>.makeStream(bufferingPolicy: .unbounded)
}
private func sdp(_ type: SDPType, restart: Bool = false) -> String { "v=0\r\ns=inmemory\r\na=type:\(type.rawValue)\r\na=handle:\(id)\r\n\(restart ? "a=restart\r\n" : "")" }
private static func handleId(in sdp: String) -> String? {
sdp.split(separator: "\r\n").first { $0.hasPrefix("a=handle:") }.map { String($0.dropFirst("a=handle:".count)) }
}
func createOffer(iceRestart: Bool) async throws -> String {
lock.withLock { state = .haveLocalOffer }
emit.yield(.iceCandidate(candidate: "candidate:1 1 udp 1 127.0.0.1 9 typ host", sdpMid: "0", sdpMLineIndex: 0))
emit.yield(.iceGatheringComplete)
return sdp(.offer, restart: iceRestart)
}
func createAnswer() async throws -> String {
let remote: String? = lock.withLock { state = .stable; return remoteId }
guard let remote else { throw P2PError.config("createAnswer without a remote offer") }
emit.yield(.iceCandidate(candidate: "candidate:1 1 udp 1 127.0.0.1 9 typ host", sdpMid: "0", sdpMLineIndex: 0))
emit.yield(.iceGatheringComplete)
emit.yield(.connectionState(.connecting))
await network.connect(remote, id)
return sdp(.answer)
}
func setRemoteDescription(_ type: SDPType, sdp: String) async throws {
guard let other = InMemoryPeerConnection.handleId(in: sdp) else { throw P2PError.config("bad in-memory SDP") }
lock.withLock {
remoteId = other
state = type == .offer ? .haveRemoteOffer : .stable
}
if type == .answer { emit.yield(.connectionState(.connecting)) }
}
func rollback() async throws { lock.withLock { state = .stable } }
func addIceCandidate(candidate: String, sdpMid: String?, sdpMLineIndex: Int?) async throws {}
func signalingState() async -> SignalingState { lock.withLock { closed ? .closed : state } }
func send(text: String) async throws {
guard lock.withLock({ open }) else { throw P2PError.transport("data channel not open") }
await network.deliver(from: id, Data(text.utf8), isBinary: false)
}
func send(data: Data) async throws {
guard lock.withLock({ open }) else { throw P2PError.transport("data channel not open") }
await network.deliver(from: id, data, isBinary: true)
}
func bufferedAmount() async -> Int { 0 }
func close() async {
let wasOpen: Bool = lock.withLock { defer { closed = true; open = false }; return open }
await network.unregister(id)
if wasOpen { emit.yield(.dataChannelClosed) }
emit.yield(.connectionState(.closed))
emit.finish()
}
// network side
func connected() {
lock.withLock { open = true }
emit.yield(.connectionState(.connected))
emit.yield(.dataChannelOpen)
}
func fail() {
let wasOpen: Bool = lock.withLock { defer { open = false }; return open }
if wasOpen { emit.yield(.dataChannelClosed) }
emit.yield(.connectionState(.failed))
}
func peerClosed() {
let wasOpen: Bool = lock.withLock { defer { open = false }; return open }
if wasOpen { emit.yield(.dataChannelClosed) }
emit.yield(.connectionState(.disconnected))
}
func receive(_ data: Data, isBinary: Bool) {
emit.yield(.message(data, isBinary: isBinary))
}
}
|