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
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574 | (ns ird.p2p.room
"One joined room: verified peers, end-to-end encrypted messaging over
WebRTC data channels with the relay as fallback, and a channel of events.
Membership is proven by cryptography, not by the roster: a peer is a
member the moment one of its envelopes decrypts under K_sig(from) with
our id as `to`. Roster peers get a hello and are ignored as strangers if
nothing decryptable arrives within the hello timeout. The lower peer id
offers a data channel through the injectable peer-connection factory;
when a channel is lost the offerer retries after 2, 5 and 8 seconds and
messages keep flowing through the relay.
Events (plain maps on `(:events room)`):
{:type :status :status {:state :joined}} / {:state :reconnecting :attempt n} / {:state :closed :reason s}
{:type :roster :peers [...]} verified members after a (re)join settles
{:type :peer-joined :peer p} a peer proved membership
{:type :peer-left :peer p}
{:type :peer-ready :peer p :via :relayed/:direct}
{:type :peer-identity :peer p :identity {:public-key :fingerprint :name}}
{:type :message :from p :message {:kind :json :value v} or {:kind :data :payload <b64>}}
{:type :relay-error :code s} the relay refused something; the room stays joined"
(:require [clojure.string :as str]
[cljs.core.async :as async :refer [go go-loop <! >!]]
[ird.p2p.async :as a :refer [go-try <?]]
[ird.p2p.crypto :as crypto]
[ird.p2p.envelope :as env]
[ird.p2p.errors :as err]
[ird.p2p.identity :as ident]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.json :as json]
[ird.p2p.impl.node :as node]
[ird.p2p.transport :as t]
[ird.p2p.webrtc :as webrtc]))
(def default-options
{:hello-timeout-ms 10000
:prefer-data-channel? true
:max-inflight-chunk-bytes env/default-max-inflight-bytes
:binary-direct-max-bytes 16000})
(def ice-restart-ladder-ms [2000 5000 8000])
(def max-pending-ice 64)
(def ^:private trace-sink (atom nil))
(defn set-trace!
"Diagnostic hook: the SDK's own tests attach a sink (fn [string]) to see
the room's internal decisions."
[f] (reset! trace-sink f))
(defn- trace [room & parts]
(when-let [f @trace-sink]
(f (str (subs (:room-id room) 0 6) "/" (subs (:peer-id room) 0 4) " " (str/join " " parts)))))
;; ---- state helpers -------------------------------------------------------------------------
(defn- new-peer [room]
{:verified? false :hello-sent? false :relay-present? true :transport :relayed :identity nil
:pc nil :pc-gen 0 :remote-desc? false :pending-ice []
:replay-sig (crypto/make-replay-window) :replay-msg (crypto/make-replay-window)
:assembler (env/make-assembler (get-in room [:options :max-inflight-chunk-bytes]))
:hello-gen 0 :restart-attempt 0 :restart-pending? false :restart-gen 0 :negotiating? false})
(defn- peer [room id] (get-in @(:state room) [:peers id]))
(defn- update-peer! [room id f & args]
(swap! (:state room) (fn [s] (if (contains? (:peers s) id) (apply update-in s [:peers id] f args) s))))
(defn- set-peer! [room id k v] (update-peer! room id assoc k v))
(defn members
"Verified members, sorted."
[room]
(->> (:peers @(:state room)) (filter (fn [[_ p]] (:verified? p))) (map key) sort vec))
(defn transport-of
"How PEER is reached (:relayed or :direct), nil when not a verified member."
[room peer-id]
(let [p (peer room peer-id)] (when (:verified? p) (:transport p))))
(defn identity-of
"PEER's verified announced identity, or nil."
[room peer-id]
(:identity (peer room peer-id)))
(defn status
"The last status map ({:state ...})."
[room]
(:status @(:state room)))
(defn closed? [room] (:closed? @(:state room)))
(defn- emit! [room event] ((:emit room) event))
(defn- emit-status! [room st]
(swap! (:state room) assoc :status st)
(emit! room {:type :status :status st}))
(defn- offerer? [room id] (neg? (compare (:peer-id room) id)))
;; ---- signaling path --------------------------------------------------------------------------
(defn- send-inner
"Seal INNER for TO and send it through the relay; failures are traced only."
[room inner to]
(go
(let [envelope (<! (env/seal-signal (:rc room) inner to))
r (<! (t/send! (:session room) {:t :signal :to to :payload envelope}))]
(when (a/error? r) (trace room "sendInner" (:kind inner) "to" (subs to 0 4) "FAILED:" (ex-message r)))
nil)))
(defn- send-hello [room id]
(go
(let [block (when-let [i (:identity room)]
(<! (ident/identity-block i (:app-salt room) (:room-id room) (:peer-id room))))]
(set-peer! room id :hello-sent? true)
(<! (send-inner room (env/hello block) id)))))
(declare maybe-negotiate schedule-restart tear-down-pc remove-peer)
(defn- arm-hello-timer [room id]
(let [gen (inc (:hello-gen (peer room id)))]
(set-peer! room id :hello-gen gen)
(go
(<! ((:sleep-fn room) (get-in room [:options :hello-timeout-ms])))
(let [p (peer room id)]
(when (and p (not (:verified? p)) (= gen (:hello-gen p)) (not (closed? room)))
(swap! (:state room) update :strangers conj id)
(<! (remove-peer room id false)))))))
(defn- ensure-peer
"Knows about ID and makes sure a hello is on its way (unless it is a known stranger)."
[room id]
(go
(let [s @(:state room)]
(when (and (not= id (:peer-id room)) (not (contains? (:strangers s) id)))
(when-not (peer room id)
(swap! (:state room) assoc-in [:peers id] (new-peer room)))
(when-not (:hello-sent? (peer room id))
(<! (send-hello room id))
(arm-hello-timer room id))))))
(defn- remove-peer [room id announce?]
(go
(when-let [p (peer room id)]
(swap! (:state room) update :peers dissoc id)
(when-let [pc (:pc p)] (<! (webrtc/close pc)))
(when announce? (emit! room {:type :peer-left :peer id}))
nil)))
(defn- relay-left [room id]
(go
(when-let [p (peer room id)]
(set-peer! room id :relay-present? false)
(if (and (:verified? p) (= :direct (:transport p)))
nil ; still talking directly; keep the peer until the data channel dies
(<! (remove-peer room id (:verified? p)))))))
(defn- verified!
"The first decrypted envelope proves membership."
[room id]
(go
(when-let [p (peer room id)]
(when-not (:hello-sent? p) (<! (send-hello room id)))
(when-not (:verified? (peer room id))
(update-peer! room id assoc :verified? true :hello-gen (inc (:hello-gen (peer room id))))
(emit! room {:type :peer-joined :peer id})
(emit! room {:type :peer-ready :peer id :via :relayed})
(<! (maybe-negotiate room id false))))))
;; ---- inner messages ---------------------------------------------------------------------------
(defn- clamp-identity
"The identity surfaced for a verified hello block: the public key, its
fingerprint and the name truncated to 32 characters (nil when absent).
Nothing else the peer announced is carried over."
[block fp]
{:public-key (:pub block)
:fingerprint fp
:name (when (string? (:name block)) (subs (:name block) 0 (min 32 (count (:name block)))))})
(declare handle-offer handle-answer handle-ice receive-direct)
(defn- emit-message! [room from message]
(emit! room {:type :message :from from :message message}))
(defn- handle-inner [room inner from]
(go
(case (:kind inner)
:hello (when-let [block (:id inner)]
(when-let [fp (<! (ident/verify-assertion (:app-salt room) (:room-id room) from (:pub block) (:sig block)))]
(let [id (clamp-identity block fp)]
(when (not= id (:identity (peer room from)))
(set-peer! room from :identity id)
(emit! room {:type :peer-identity :peer from :identity id})))))
:offer (<! (handle-offer room (:sdp inner) from))
:answer (<! (handle-answer room (:sdp inner) from))
:ice (<! (handle-ice room (:candidate inner) (:sdp-mid inner) (:sdp-mline-index inner) from))
:restart (when (offerer? room from) (schedule-restart room from))
:msg (emit-message! room from {:kind :json :value (:payload inner)})
:bin (emit-message! room from {:kind :data :payload (:data inner)})
:chunk (when (peer room from)
(let [result (atom nil)]
(update-peer! room from update :assembler
(fn [asm] (let [[asm' r] (env/assemble asm inner ((:now-fn room)))] (reset! result r) asm')))
(when (map? @result)
(let [{:keys [enc data]} @result]
(case enc
:bin (emit-message! room from {:kind :data :payload data})
:json (when-let [m (some-> (bytes/b64->utf8 data) json/parse env/parse-inner)]
(when (= :msg (:kind m)) (emit-message! room from {:kind :json :value (:payload m)}))))))))
nil)))
(defn- handle-signal [room from payload]
(go
(when (not= from (:peer-id room))
(if-let [envelope (env/parse-envelope payload)]
(if-let [plain (<! (env/open-signal (:rc room) envelope from))]
(do
(trace room "got" (:kind plain) "from" (subs from 0 4))
(when-not (peer room from)
(swap! (:state room) (fn [s] (-> s (update :strangers disj from) (assoc-in [:peers from] (new-peer room))))))
(let [accepted (atom false)]
(update-peer! room from update :replay-sig
(fn [w] (let [[w' ok] (crypto/replay-accept w (:iv envelope))] (reset! accepted ok) w')))
(when @accepted
(<! (verified! room from))
(when-let [inner (env/parse-inner plain)]
(<! (handle-inner room inner from))))))
(trace room "drop undecryptable from" (subs from 0 4)))
(trace room "drop non-envelope from" (subs from 0 4))))))
;; ---- WebRTC negotiation (lower peer id offers; higher answers) ---------------------------------
(defn- attach [room pc id]
(let [gen (inc (:pc-gen (peer room id)))]
(update-peer! room id assoc :pc pc :pc-gen gen :remote-desc? false :pending-ice [])
(go-loop []
(when-let [ev (<! (webrtc/events pc))]
(<! ((:handle-pc-event room) room ev id gen))
(recur)))))
(defn- tear-down-pc [room id]
(go
(when-let [p (peer room id)]
(let [old (:pc p)]
(update-peer! room id assoc :pc nil :remote-desc? false :pending-ice [] :pc-gen (inc (:pc-gen p)))
(when (= :direct (:transport p))
(set-peer! room id :transport :relayed)
(when (:verified? p) (emit! room {:type :peer-ready :peer id :via :relayed})))
(when old (<! (webrtc/close old)))))))
(defn- maybe-negotiate [room id restart?]
(go
(let [p (peer room id)]
(when (and (get-in room [:options :prefer-data-channel?]) (offerer? room id) p (:verified? p) (:relay-present? p)
(not (and (= :direct (:transport p)) (not restart?)))
(not (:negotiating? p)))
(set-peer! room id :negotiating? true)
(<! (tear-down-pc room id))
(let [ice (<! ((:ice-fn room)))
pc (<! (webrtc/make-peer-connection (:pc-factory room) ice))]
(if (a/error? pc)
(do (set-peer! room id :negotiating? false) (schedule-restart room id))
(do
(attach room pc id)
(let [sdp (<! (webrtc/create-offer pc restart?))]
(if (a/error? sdp)
(do (set-peer! room id :negotiating? false) (schedule-restart room id))
(<! (send-inner room (env/offer sdp) id)))))))))))
(defn- flush-ice [room id]
(go
(when-let [p (peer room id)]
(when-let [pc (:pc p)]
(let [pending (:pending-ice p)]
(set-peer! room id :pending-ice [])
(doseq [[c m i] pending] (<! (webrtc/add-ice-candidate pc c m i))))))))
(defn- handle-offer [room sdp from]
(go
(let [p (peer room from)]
(if (and (get-in room [:options :prefer-data-channel?]) (not (offerer? room from)) (:verified? p))
(do
(<! (tear-down-pc room from))
(let [ice (<! ((:ice-fn room)))
pc (<! (webrtc/make-peer-connection (:pc-factory room) ice))]
(if (a/error? pc)
(trace room "handleOffer failed:" (ex-message pc))
(do
(attach room pc from)
(let [r (<! (webrtc/set-remote-description pc :offer sdp))]
(if (a/error? r)
(do (trace room "handleOffer failed:" (ex-message r)) (<! (tear-down-pc room from)))
(do
(set-peer! room from :remote-desc? true)
(<! (flush-ice room from))
(let [answer (<! (webrtc/create-answer pc))]
(if (a/error? answer)
(do (trace room "handleOffer failed:" (ex-message answer)) (<! (tear-down-pc room from)))
(<! (send-inner room (env/answer answer) from)))))))))))
(trace room "handleOffer guard failed")))))
(defn- handle-answer [room sdp from]
(go
(let [p (peer room from)]
(when (and (offerer? room from) (:pc p) (not (:remote-desc? p)))
(let [r (<! (webrtc/set-remote-description (:pc p) :answer sdp))]
(if (a/error? r)
(do (set-peer! room from :negotiating? false) (schedule-restart room from))
(do (set-peer! room from :remote-desc? true)
(<! (flush-ice room from)))))))))
(defn- handle-ice [room candidate mid index from]
(go
(when candidate ; end-of-candidates carries nothing to add
(when-let [p (peer room from)]
(if (and (:remote-desc? p) (:pc p))
(<! (webrtc/add-ice-candidate (:pc p) candidate mid index))
(when (< (count (:pending-ice p)) max-pending-ice)
(update-peer! room from update :pending-ice conj [candidate mid index])))))))
(defn- schedule-restart
"The next rung of the ICE-restart ladder. Idempotent while a restart is
already pending: a lost data channel reports several events."
[room id]
(let [p (peer room id)]
(when (and p (:verified? p) (:relay-present? p) (not (:restart-pending? p))
(< (:restart-attempt p) (count ice-restart-ladder-ms)))
(let [delay (nth ice-restart-ladder-ms (:restart-attempt p))
gen (:restart-gen p)]
(update-peer! room id assoc :restart-pending? true :restart-attempt (inc (:restart-attempt p)))
(go
(<! ((:sleep-fn room) delay))
(let [p (peer room id)]
(when (and p (= gen (:restart-gen p)) (not (closed? room)))
(set-peer! room id :restart-pending? false)
(<! (maybe-negotiate room id true)))))))
nil))
(defn- handle-pc-event [room {:keys [type] :as ev} id gen]
(go
(let [p (peer room id)]
(if-not (and p (= gen (:pc-gen p)))
(trace room "pc event" type "for" (subs id 0 4) "ignored")
(do
(trace room "pc event" type "from" (subs id 0 4))
(case type
:ice-candidate (<! (send-inner room (env/ice (:candidate ev) (:sdp-mid ev) (:sdp-mline-index ev)) id))
:ice-gathering-complete (<! (send-inner room (env/ice nil nil nil) id))
:data-channel-open
(do (update-peer! room id assoc :transport :direct :negotiating? false :restart-attempt 0
:restart-pending? false :restart-gen (inc (:restart-gen p)))
(when (:verified? p) (emit! room {:type :peer-ready :peer id :via :direct})))
(:data-channel-closed :connection-state)
(when (or (= type :data-channel-closed) (contains? #{:failed :closed :disconnected} (:state ev)))
(let [was-direct (= :direct (:transport p))]
(update-peer! room id assoc :transport :relayed :negotiating? false)
(if-not (:relay-present? p)
(<! (remove-peer room id (:verified? p)))
(do (when (and was-direct (:verified? p)) (emit! room {:type :peer-ready :peer id :via :relayed}))
(when (offerer? room id) (schedule-restart room id))))))
:message (<! (receive-direct room (:data ev) (:binary? ev) id))
nil))))))
;; ---- data path ------------------------------------------------------------------------------------
(defn- check! [v] (when (a/error? v) (throw v)) v)
(defn- send-direct [room message id pc]
(go-try
(let [rc (:rc room)
slice env/data-channel-slice-size
send-env (fn [inner] (go-try (check! (<! (webrtc/send-text pc (env/envelope->json (<! (env/seal-message rc inner))))))))]
(case (:kind message)
:data (let [data (:payload message)]
(if (<= (bytes/byte-length data) (get-in room [:options :binary-direct-max-bytes]))
(check! (<! (webrtc/send-binary pc (<! (env/seal-message-binary rc data)))))
(doseq [c (env/split data "bin" slice)] (<? (send-env c)))))
:json (let [inner (env/msg (:value message))
b64 (bytes/utf8->b64 (json/stringify inner))]
(if (<= (bytes/byte-length b64) slice)
(<? (send-env inner))
(doseq [c (env/split b64 "json" slice)] (<? (send-env c))))))
true)))
(defn- send-relayed [room message id]
(go-try
(let [rc (:rc room)
slice (env/relay-slice-size (:max-frame-bytes room))
send-env (fn [inner]
(go-try (check! (<! (t/send! (:session room) {:t :signal :to id :payload (<! (env/seal-signal rc inner id))})))))]
(case (:kind message)
:data (let [data (:payload message)]
(if (<= (bytes/byte-length data) slice)
(<? (send-env (env/bin data)))
(doseq [c (env/split data "bin" slice)] (<? (send-env c)))))
:json (let [inner (env/msg (:value message))
b64 (bytes/utf8->b64 (json/stringify inner))]
(if (<= (bytes/byte-length b64) slice)
(<? (send-env inner))
(doseq [c (env/split b64 "json" slice)] (<? (send-env c))))))
true)))
(defn- valid-message? [m]
(and (map? m)
(case (:kind m)
:json (contains? m :value)
:data (bytes/valid-b64? (:payload m))
false)))
(defn send
"Send MESSAGE ({:kind :json :value v} or {:kind :data :payload <b64>}) to
one verified peer: over its data channel when open, else through the
relay. Channel of true, or :ird.p2p/closed, :ird.p2p/peer-unknown,
:ird.p2p/config-error (malformed message), :ird.p2p/frame-too-large."
[room to message]
(go-try
(when (closed? room) (throw (err/closed)))
(when-not (valid-message? message)
(throw (err/config-error "message must be {:kind :json :value v} or {:kind :data :payload <base64>}" {:message message})))
(let [p (peer room to)]
(when-not (:verified? p) (throw (err/peer-unknown to)))
(if (and (= :direct (:transport p)) (:pc p) (get-in room [:options :prefer-data-channel?]))
(let [r (<! (send-direct room message to (:pc p)))]
(if (a/error? r)
(<? (send-relayed room message to)) ; the channel closed mid-send
r))
(<? (send-relayed room message to))))))
(defn broadcast
"Send MESSAGE to every verified peer; failures for individual peers are
swallowed. Channel of true."
[room message]
(go
(doseq [p (members room)] (<! (send room p message)))
true))
(defn- receive-direct [room data binary? id]
(go
(when (:verified? (peer room id))
(if binary?
(when (and (bytes/valid-b64? data) (>= (bytes/byte-length data) env/minimum-length))
(let [iv (bytes/slice-b64 data 1 13)
accepted (atom false)]
(update-peer! room id update :replay-msg
(fn [w] (let [[w' ok] (crypto/replay-accept w iv)] (reset! accepted ok) w')))
(when @accepted
(when-let [plain (<! (env/open-message-binary (:rc room) data id))]
(emit-message! room id {:kind :data :payload plain})))))
(when-let [envelope (env/json->envelope data)]
(let [accepted (atom false)]
(update-peer! room id update :replay-msg
(fn [w] (let [[w' ok] (crypto/replay-accept w (:iv envelope))] (reset! accepted ok) w')))
(when @accepted
(when-let [plain (<! (env/open-message (:rc room) envelope id))]
(when-let [inner (env/parse-inner plain)]
(<! (handle-inner room inner id)))))))))))
;; ---- relay events ----------------------------------------------------------------------------------
(defn- close-everything! [room reason]
(go
(when-not (closed? room)
(swap! (:state room) assoc :closed? true)
(let [peers (:peers @(:state room))]
(swap! (:state room) assoc :peers {})
(doseq [[_ p] peers] (when-let [pc (:pc p)] (<! (webrtc/close pc)))))
(emit-status! room {:state :closed :reason reason})
((:close-events room)))))
(defn- handle-relay [room {:keys [t] :as frame}]
(go
(when-not (closed? room)
(case t
:joined (emit-status! room {:state :joined})
:peers (let [fresh (set (:peers frame))
s @(:state room)]
;; peers gone from the roster while we were away
(doseq [[id p] (:peers s)]
(when (and (not (contains? fresh id)) (:relay-present? p)) (<! (relay-left room id))))
(swap! (:state room) assoc :relay-roster fresh)
(doseq [id (:peers frame)] (<! (ensure-peer room id)))
(when (:joined-once? s)
;; rejoin: re-hello everyone so peers with a fresh session re-verify us
(doseq [id (:peers frame)] (when (:verified? (peer room id)) (<! (send-hello room id)))))
(swap! (:state room) assoc :joined-once? true)
(let [roster (members room)]
(swap! (:state room) assoc :roster roster)
(emit! room {:type :roster :peers roster})))
:peer-joined (let [id (:peer frame)]
(swap! (:state room) (fn [s] (-> s (update :relay-roster conj id) (update :strangers disj id))))
(if (peer room id)
(do (set-peer! room id :relay-present? true)
(<! (send-hello room id)))
(<! (ensure-peer room id))))
:peer-left (let [id (:peer frame)]
(swap! (:state room) update :relay-roster disj id)
(<! (relay-left room id)))
:signal (<! (handle-signal room (:from frame) (:payload frame)))
:error (emit! room {:type :relay-error :code (:code frame)})
:reconnecting (emit-status! room {:state :reconnecting :attempt (:attempt frame)})
:closed (<! (close-everything! room (:reason frame)))
nil))))
;; ---- public surface ---------------------------------------------------------------------------------
(defn wake
"The app came to the foreground or the network changed: reconnect now,
re-hello silent peers and retry lost data channels. Channel."
[room]
(go
(when-not (closed? room)
(t/wake! (:session room))
(swap! (:state room) assoc :strangers #{})
(doseq [id (:relay-roster @(:state room))]
(when-not (:verified? (peer room id)) (<! (ensure-peer room id))))
(doseq [[id p] (:peers @(:state room))]
(when (and (:verified? p) (= :relayed (:transport p)) (:relay-present? p))
(update-peer! room id assoc :restart-pending? false :restart-attempt 0 :restart-gen (inc (:restart-gen p)))
(<! (maybe-negotiate room id true))))
true)))
(defn leave
"Leave the room and close every connection; the room cannot be rejoined.
Channel that closes when done."
[room]
(go
(when-not (closed? room)
(swap! (:state room) assoc :closed? true)
(let [peers (:peers @(:state room))]
(swap! (:state room) assoc :peers {})
(doseq [[_ p] peers] (when-let [pc (:pc p)] (<! (webrtc/close pc)))))
(<! (t/close! (:session room)))
(emit-status! room {:state :closed :reason "left"})
((:close-events room)))
nil))
(defn join
"Join a room. OPTS:
:rc the room crypto context (ird.p2p.crypto/room-crypto)
:transport an ird.p2p.transport/Transport
:transport-opts its connect options (:url :origin :limits ...; :room/:peer are filled in)
:pc-factory an ird.p2p.webrtc/PeerConnectionFactory
:ice-fn (fn [] → channel of the ICE configuration)
:identity an ird.p2p.identity map, or nil
:app-salt the app salt
:max-frame-bytes the relay's frame cap
:options room options (see `default-options`)
:sleep-fn :now-fn
Channel of the room map (events on (:events room)) or the error that
refused the join."
[{:keys [rc transport transport-opts pc-factory ice-fn identity app-salt max-frame-bytes options sleep-fn now-fn]}]
(go-try
(let [{:keys [out put! close!]} (a/unbounded-chan)
session (<? (t/connect transport (assoc transport-opts :room (:room-id rc) :peer (:peer-id rc))))
room {:room-id (:room-id rc)
:peer-id (:peer-id rc)
:app-salt app-salt
:rc rc
:session session
:identity identity
:options (merge default-options options)
:pc-factory pc-factory
:ice-fn (or ice-fn (fn [] (go {:servers [] :relay-only? false})))
:max-frame-bytes max-frame-bytes
:sleep-fn (or sleep-fn a/sleep)
:now-fn (or now-fn node/now-ms)
:events out
:emit put!
:close-events close!
:handle-pc-event handle-pc-event
:state (atom {:closed? false :joined-once? false :peers {} :strangers #{} :relay-roster #{}
:status {:state :connecting} :roster []})}]
(go-loop []
(when-let [frame (<! (t/inbound session))]
(<! (handle-relay room frame))
(recur)))
room)))
|