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 | import Foundation
import IrdP2PCrypto
import IrdP2PSignaling
/// How a peer is currently reached.
public enum PeerTransport: Sendable, Equatable {
/// Through the relay's `signal` frames (paced, size-capped).
case relayed
/// Over a WebRTC data channel.
case direct
}
public enum ConnectionStatus: Sendable, Equatable {
case connecting
case joined
case reconnecting(attempt: Int)
case closed(reason: String)
}
/// An application message: JSON or raw bytes, end-to-end encrypted in transit.
public enum Message: Sendable, Equatable {
case json(JSONValue)
case data(Data)
public static func json<T: Encodable>(_ value: T) throws -> Message { .json(try JSONValue(encoding: value)) }
/// Decodes a `.json` message; throws for `.data`.
public func decode<T: Decodable>(_ type: T.Type) throws -> T {
guard case .json(let v) = self else { throw P2PError.config("message is binary, not JSON") }
return try v.decode(type)
}
public var jsonValue: JSONValue? { if case .json(let v) = self { return v }; return nil }
public var dataValue: Data? { if case .data(let d) = self { return d }; return nil }
}
/// A verified peer's announced identity.
public struct PeerIdentity: Sendable, Equatable {
public var publicKeyBase64URL: String
public var fingerprint: Fingerprint
public var name: String?
public var hue: Int?
public var glyph: String?
}
/// Why a partially reassembled message was given up.
public enum AssemblyDropReason: Sendable, Equatable {
/// Still incomplete after `RoomOptions.maxChunkAge`.
case ageExpired
/// A new id arrived while 8 assemblies were already in flight for the peer.
case tooManyConcurrent
/// The slice would exceed `RoomOptions.maxInflightChunkBytes`.
case tooManyBytes
/// The slice disagrees with its assembly (count or encoding) or its index is out of range.
case mismatched
/// The slice repeats an index with different bytes.
case duplicateIndex
}
public enum RoomEvent: Sendable, Equatable {
case status(ConnectionStatus)
/// The verified members after a (re)join settles.
case roster([PeerId])
/// A peer proved membership (its first envelope decrypted).
case peerJoined(PeerId)
case peerLeft(PeerId)
/// The peer's transport changed: `.relayed` at verification, `.direct` when the data channel opens, back on loss.
case peerReady(PeerId, PeerTransport)
case peerIdentity(PeerId, PeerIdentity)
case message(from: PeerId, Message)
/// The relay refused something (`rateLimited`, ...). The room stays joined.
case relayError(RelayErrorCode)
/// A chunked message from `peer` was given up before it completed; `id` is its chunk id.
/// Age expiry is noticed when the peer's next chunk arrives.
case assemblyDropped(peer: PeerId, id: String, reason: AssemblyDropReason)
}
/// Fan-out of events to any number of `AsyncStream` subscribers. "Sticky" events
/// (the latest status and roster) are replayed to every new subscriber first, so
/// code that attaches after the room started still learns the current state.
final class EventBus<Event: Sendable>: @unchecked Sendable {
private let lock = NSLock()
private var subscribers: [UUID: AsyncStream<Event>.Continuation] = [:]
private var finished = false
private var stickyOrder: [String] = []
private var sticky: [String: Event] = [:]
private let policy: AsyncStream<Event>.Continuation.BufferingPolicy
init(policy: AsyncStream<Event>.Continuation.BufferingPolicy = .unbounded) { self.policy = policy }
func stream() -> AsyncStream<Event> {
let id = UUID()
let (s, c) = AsyncStream<Event>.makeStream(bufferingPolicy: policy)
let (done, replay): (Bool, [Event]) = lock.withLock {
if finished { return (true, []) }
subscribers[id] = c
return (false, stickyOrder.compactMap { sticky[$0] })
}
for e in replay { c.yield(e) }
if done { c.finish(); return s }
c.onTermination = { [weak self] _ in
guard let self else { return }
_ = self.lock.withLock { self.subscribers.removeValue(forKey: id) }
}
return s
}
func send(_ event: Event) {
let subs = lock.withLock { Array(subscribers.values) }
for c in subs { c.yield(event) }
}
/// Sends and remembers the event under `key`; a later subscriber receives the latest one per key first.
func sendSticky(_ key: String, _ event: Event) {
let subs: [AsyncStream<Event>.Continuation] = lock.withLock {
if sticky[key] == nil { stickyOrder.append(key) }
sticky[key] = event
return Array(subscribers.values)
}
for c in subs { c.yield(event) }
}
func finish() {
let subs: [AsyncStream<Event>.Continuation] = lock.withLock {
finished = true
let v = Array(subscribers.values)
subscribers.removeAll()
return v
}
for c in subs { c.finish() }
}
}
|