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 | (ns ird.p2p.fake-relay
"An in-process relay with the `/ws` hub semantics: one join per socket,
rooms namespaced by id, `from` stamped by the relay, unknown recipients
dropped, rejoin-with-same-id replaces the old socket, `ping` answered
with `pong`. Exposed as a :socket-fn for ird.p2p.transport.ws, so the real
session logic runs over it."
(:require [cljs.core.async :as async :refer [go]]
[ird.p2p.async :as a]
[ird.p2p.crypto :as crypto]
[ird.p2p.errors :as err]
[ird.p2p.impl.json :as json]))
(defn make-relay
"OPTS: :max-room-peers (16), :reject-with (an HTTP status to refuse every
upgrade with), :duplicate-signals? (every forwarded signal is delivered
twice), :drop-next-signals (atom count)."
([] (make-relay nil))
([{:keys [max-room-peers reject-with]}]
{:rooms (atom {})
:frames (atom [])
:max-room-peers (or max-room-peers 16)
:reject-with (atom reject-with)
:duplicate-signals? (atom false)
:drop-next-signals (atom 0)
:connections (atom 0)}))
(defn- deliver-frame [sock frame] ((:put! sock) {:type :text :data (json/stringify frame)}))
(defn- terminate! [sock]
(when (compare-and-set! (:closed? sock) false true)
((:close-in! sock))))
(defn- disconnect! [relay sock]
(let [{:keys [room peer]} @(:joined sock)]
(if (and room peer)
(let [others (atom [])]
(swap! (:rooms relay)
(fn [rooms]
(if (identical? (get-in rooms [room peer]) sock)
(let [members (dissoc (get rooms room) peer)]
(reset! others (vals members))
(if (empty? members) (dissoc rooms room) (assoc rooms room members)))
rooms)))
(terminate! sock)
(doseq [o @others] (deliver-frame o {:t "peer-left" :peer peer})))
(terminate! sock))))
(defn- handle! [relay sock text]
(let [v (json/parse text)]
(when (and (map? v) (string? (:t v)))
(swap! (:frames relay) conj {:from (:peer @(:joined sock)) :text text})
(case (:t v)
"join"
(let [room (:room v) peer (:peer v)]
(if-not (and (nil? (:room @(:joined sock))) (crypto/valid-room-id? room) (crypto/valid-peer-id? peer))
(deliver-frame sock {:t "error" :code "bad-message"})
(let [replaced (atom nil) others (atom nil) full (atom false)]
(swap! (:rooms relay)
(fn [rooms]
(let [members (get rooms room {})
old (get members peer)]
(if (and (nil? old) (>= (count members) (:max-room-peers relay)))
(do (reset! full true) rooms)
(do (reset! replaced old)
(reset! others (vals (dissoc members peer)))
(assoc rooms room (assoc members peer sock)))))))
(if @full
(deliver-frame sock {:t "error" :code "room-full"})
(do
(reset! (:joined sock) {:room room :peer peer})
(when @replaced (terminate! @replaced))
(deliver-frame sock {:t "peers" :peers (vec (keep #(:peer @(:joined %)) @others))})
(doseq [o @others] (deliver-frame o {:t "peer-joined" :peer peer})))))))
"signal"
(let [{:keys [room peer]} @(:joined sock)]
(cond
(not (and room peer)) (deliver-frame sock {:t "error" :code "not-joined"})
(not (and (crypto/valid-peer-id? (:to v)) (contains? v :payload))) (deliver-frame sock {:t "error" :code "bad-message"})
(pos? @(:drop-next-signals relay)) (swap! (:drop-next-signals relay) dec)
:else (when-let [target (get-in @(:rooms relay) [room (:to v)])]
(let [frame {:t "signal" :from peer :payload (:payload v)}]
(deliver-frame target frame)
(when @(:duplicate-signals? relay) (deliver-frame target frame))))))
"leave" (disconnect! relay sock)
"ping" (deliver-frame sock {:t "pong"})
(deliver-frame sock {:t "error" :code "bad-message"})))))
(defn socket-fn
"The :socket-fn of RELAY."
[relay]
(fn [_opts]
(let [out (async/promise-chan)]
(if-let [status @(:reject-with relay)]
(async/put! out (err/upgrade-rejected status))
(let [{:keys [put! close!] in :out} (a/unbounded-chan)
sock {:put! put! :close-in! close! :joined (atom {}) :closed? (atom false)}]
(swap! (:connections relay) inc)
(async/put! out {:in in
:send! (fn [text]
(if @(:closed? sock)
(err/closed)
(do (handle! relay sock text) nil)))
:close! (fn [_code _reason] (disconnect! relay sock) nil)})))
out)))
(defn drop-socket!
"Kill PEER's socket without a leave (network loss)."
[relay peer]
(when-let [sock (some (fn [[_ members]] (get members peer)) @(:rooms relay))]
(disconnect! relay sock)))
(defn members [relay room] (sort (keys (get @(:rooms relay) room))))
(defn signal-frames [relay]
(filter #(re-find #"^\{\"t\":\"signal\"" (:text %)) @(:frames relay)))
|