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 | (ns ird.p2p.transport.libp2p
"A relay transport behind the same SPI as ird.p2p.transport.ws, over real
libp2p: the room is a gossipsub topic (derived from the room id), directed
signal frames are reconstructed over pubsub by carrying :from and :to in
the published payload — so the envelope AAD `v1|roomId|from|to|sig` stays
exact — and presence announcements give the room the peer-joined/peer-left
and roster events it needs to start the hello/verify handshake. Sender ids
are the 32-lowercase-hex relay peer ids the room supplies, published
in-band, so K_sig(from) works identically to the /ws relay.
The libp2p node is injectable through :node-fn (default
ird.p2p.impl.libp2p/create-node), so this namespace stays pure: it speaks
only the node handle's data and channels. The offline tests inject an
in-process pubsub bus; the real two-node path over loopback tcp is
always on (ird.p2p.libp2p-node-test; IRD_P2P_NO_LIBP2P_NODE=1 skips it)."
(:require [clojure.string :as str]
[cljs.core.async :as async :refer [go go-loop <! >! alts!]]
[ird.p2p.async :as a :refer [go-try <?]]
[ird.p2p.crypto :as crypto]
[ird.p2p.errors :as err]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.json :as json]
[ird.p2p.impl.libp2p :as impl]
[ird.p2p.impl.node :as node]
[ird.p2p.transport :as t]))
(defn room-topic
"The gossipsub topic of a room: `ap2p/room/<roomId>`. Both peers derive
the same room id, so they meet on the same topic."
[room-id]
(str "ap2p/room/" room-id))
(defn- make-session
[{:keys [node topic peer own-node? relay-multiaddrs]}]
(let [{:keys [out put! close!]} (a/unbounded-chan)
st (atom :joined)
peers (atom #{})
publish (fn [frame] ((:publish node) topic (bytes/utf8->b64 (json/stringify frame))))
note-peer! (fn [from]
(when (and from (not= from peer) (not (contains? @peers from)))
(swap! peers conj from)
(put! {:t :peer-joined :peer from})
;; re-announce so a peer that joined after us learns our presence too
(publish {:t "presence" :from peer})))
session
(reify t/Session
(send! [_ frame]
(go-try
(case (:t frame)
:signal (do (<! (publish {:t "signal" :from peer :to (:to frame) :payload (:payload frame)})) true)
:ping true
:leave true
true)))
(inbound [_] out)
(state [_] @st)
(wake! [_] (publish {:t "presence" :from peer}) nil)
(close! [_]
(go
(when (not= :closed @st)
(reset! st :closed)
(<! (publish {:t "bye" :from peer}))
((:unsubscribe node) topic)
(when own-node? (<! ((:stop node))))
(put! {:t :closed :reason "left"})
(close!))
nil)))]
;; message loop
(go-loop []
(when-let [m (<! (:messages node))]
(when (and (= topic (:topic m)) (not= :closed @st))
(when-let [frame (some-> (bytes/b64->utf8 (:data m)) json/parse)]
(case (:t frame)
"presence" (note-peer! (:from frame))
"bye" (let [f (:from frame)]
(when (contains? @peers f)
(swap! peers disj f)
(put! {:t :peer-left :peer f})))
"signal" (do (note-peer! (:from frame))
(when (= peer (:to frame))
(put! {:t :signal :from (:from frame) :payload (:payload frame)})))
nil)))
(recur)))
;; announce ourselves and settle the (initially empty) roster. gossipsub
;; drops a publish that reaches no meshed peer, and a mesh takes a moment
;; to form after two peers connect, so re-announce presence until we have
;; seen a peer (or give up after ~5 s). note-peer! also re-announces on
;; every newly seen peer, so late joiners always learn us.
(go
((:subscribe node) topic)
(put! {:t :joined})
(put! {:t :peers :peers []})
(loop [i 0]
(when (and (not= :closed @st) (empty? @peers) (< i 25))
(<! (publish {:t "presence" :from peer}))
(<! (async/timeout 200))
(recur (inc i)))))
session))
(defn transport
"A libp2p relay transport. OPTS: :node-fn (fn [{:listen :relay-multiaddrs}]
-> channel of a node handle; default ird.p2p.impl.libp2p/create-node),
:node (an already-created shared node handle to reuse instead of creating
one), :listen (multiaddrs)."
([] (transport nil))
([{:keys [node-fn node listen]}]
(let [node-fn (or node-fn impl/create-node)]
(reify t/Transport
(connect [_ opts]
(go-try
(when-not (crypto/valid-peer-id? (:peer opts))
(throw (err/config-error "libp2p transport needs a 32-hex :peer" {:peer (:peer opts)})))
(let [own? (nil? node)
relay-multiaddrs (get-in opts [:descriptor :relay :multiaddrs])
n (if node node (<? (node-fn {:listen listen :relay-multiaddrs relay-multiaddrs})))]
(make-session {:node n :topic (room-topic (:room opts)) :peer (:peer opts)
:own-node? own? :relay-multiaddrs relay-multiaddrs}))))))))
|