1// A thin WebRTC wrapper, distilled from trystero's peer.ts. We only need a
2// reliable ordered data channel plus offer/answer/ICE signaling. To keep things
3// simple we avoid "perfect negotiation" glare handling by ensuring only ONE
4// side (a deterministically chosen initiator) ever creates the offer.
5//
6// Unlike commonview's original, the channel carries two kinds of frames:
7// strings (JSON control messages) and ArrayBuffers (chunks of large binary
8// payloads, e.g. sample sets). The wrapper keeps them apart and applies
9// backpressure when streaming binary data.
11export type Signal =
12 | {type: 'offer'; sdp: string}
13 | {type: 'answer'; sdp: string}
14 | {type: 'candidate'; candidate: RTCIceCandidateInit}
16export interface PeerHandlers {
17 signal: (signal: Signal) => void
18 connect: () => void
19 data: (data: string) => void
20 binary: (data: ArrayBuffer) => void
21 close: () => void
22}
24const ICE_SERVERS: RTCIceServer[] = [
25 {urls: 'stun:stun.l.google.com:19302'},
26 {urls: 'stun:stun1.l.google.com:19302'},
27 {urls: 'stun:stun.cloudflare.com:3478'}
28]
30// Keep binary frames well under the ~256 KB cross-browser SCTP message limit.
31export const BINARY_CHUNK_BYTES = 64 * 1024
33// While streaming a large payload, pause whenever this much is queued in the
34// channel and resume once it drains below the low-water mark.
35const HIGH_WATER = 1 << 20 // 1 MiB
36const LOW_WATER = 1 << 18 // 256 KiB
38export class Peer {
39 private pc: RTCPeerConnection
40 private channel: RTCDataChannel | null = null
41 private handlers: Partial<PeerHandlers> = {}
42 private pendingCandidates: RTCIceCandidateInit[] = []
43 private closed = false
45 constructor(private initiator: boolean) {
46 this.pc = new RTCPeerConnection({iceServers: ICE_SERVERS})
48 this.pc.onicecandidate = ({candidate}) => {
49 if (candidate) {
50 this.handlers.signal?.({type: 'candidate', candidate: candidate.toJSON()})
51 }
52 }
54 this.pc.onconnectionstatechange = () => {
55 const s = this.pc.connectionState
56 if (s === 'failed' || s === 'closed' || s === 'disconnected') {
57 this.destroy()
58 }
59 }
61 if (initiator) {
62 this.setupChannel(this.pc.createDataChannel('data'))
63 this.pc.onnegotiationneeded = () => void this.makeOffer()
64 } else {
65 this.pc.ondatachannel = ({channel}) => this.setupChannel(channel)
66 }
67 }
69 setHandlers(handlers: Partial<PeerHandlers>) {
70 Object.assign(this.handlers, handlers)
71 }
73 private setupChannel(channel: RTCDataChannel) {
74 this.channel = channel
75 channel.binaryType = 'arraybuffer'
76 channel.bufferedAmountLowThreshold = LOW_WATER
77 channel.onopen = () => this.handlers.connect?.()
78 channel.onclose = () => this.destroy()
79 channel.onmessage = e => {
80 if (typeof e.data === 'string') this.handlers.data?.(e.data)
81 else this.handlers.binary?.(e.data as ArrayBuffer)
82 }
83 }
85 private async makeOffer() {
86 if (this.closed) return
87 try {
88 await this.pc.setLocalDescription(await this.pc.createOffer())
89 this.handlers.signal?.({
90 type: 'offer',
91 sdp: this.pc.localDescription!.sdp
92 })
93 } catch {
94 /* ignore */
95 }
96 }
98 async signal(signal: Signal) {
99 if (this.closed) return
100 try {
101 if (signal.type === 'candidate') {
102 if (this.pc.remoteDescription) {
103 await this.pc.addIceCandidate(signal.candidate)
104 } else {
105 this.pendingCandidates.push(signal.candidate)
106 }
107 return
108 }
110 if (signal.type === 'offer') {
111 if (this.initiator) return // initiators never accept remote offers
112 await this.pc.setRemoteDescription({type: 'offer', sdp: signal.sdp})
113 await this.flushCandidates()
114 await this.pc.setLocalDescription(await this.pc.createAnswer())
115 this.handlers.signal?.({
116 type: 'answer',
117 sdp: this.pc.localDescription!.sdp
118 })
119 return
120 }
122 if (signal.type === 'answer') {
123 await this.pc.setRemoteDescription({type: 'answer', sdp: signal.sdp})
124 await this.flushCandidates()
125 }
126 } catch {
127 /* ignore transient signaling errors */
128 }
129 }
131 private async flushCandidates() {
132 const queued = this.pendingCandidates.splice(0)
133 for (const c of queued) {
134 try {
135 await this.pc.addIceCandidate(c)
136 } catch {
137 /* ignore */
138 }
139 }
140 }
142 send(data: string) {
143 if (this.channel?.readyState === 'open') this.channel.send(data)
144 }
146 /** Stream a large binary payload as sequential chunks, respecting channel
147 * backpressure. Resolves when everything is handed to the channel; resolves
148 * false if the channel closed part-way. */
149 async sendBinary(payload: ArrayBuffer): Promise<boolean> {
150 for (let off = 0; off < payload.byteLength; off += BINARY_CHUNK_BYTES) {
151 const ch = this.channel
152 if (!ch || ch.readyState !== 'open') return false
153 if (ch.bufferedAmount > HIGH_WATER) {
154 const ok = await this.drain(ch)
155 if (!ok) return false
156 }
157 try {
158 ch.send(payload.slice(off, off + BINARY_CHUNK_BYTES))
159 } catch {
160 return false
161 }
162 }
163 return true
164 }
166 private drain(ch: RTCDataChannel): Promise<boolean> {
167 return new Promise(resolve => {
168 const done = (ok: boolean) => {
169 ch.removeEventListener('bufferedamountlow', onLow)
170 ch.removeEventListener('close', onClose)
171 resolve(ok)
172 }
173 const onLow = () => done(true)
174 const onClose = () => done(false)
175 ch.addEventListener('bufferedamountlow', onLow)
176 ch.addEventListener('close', onClose)
177 if (ch.readyState !== 'open') done(false)
178 })
179 }
181 get isConnected(): boolean {
182 return this.channel?.readyState === 'open'
183 }
185 destroy() {
186 if (this.closed) return
187 this.closed = true
188 try {
189 this.channel?.close()
190 } catch {
191 /* ignore */
192 }
193 try {
194 this.pc.close()
195 } catch {
196 /* ignore */
197 }
198 this.handlers.close?.()
199 }
200}