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 | (ns ird.p2p.impl.wrtc
"A real WebRTC peer-connection factory for Node over
`node-datachannel` (libdatachannel bindings with prebuilt binaries):
`wrtc-factory` implements ird.p2p.webrtc/PeerConnectionFactory, so an
application passes `(wrtc-factory)` as :pc-factory to ird.p2p/make-app
and rooms open genuine data channels instead of staying relayed.
Every connection owns the SPI's one negotiated data channel (id 0, label
\"ird\", ordered, reliable), created on both sides before any description
is exchanged; auto-negotiation is off so the room drives offer/answer.
The ICE configuration the room supplies ({:servers [{:urls [...]
:username :credential}] :relay-only? bool}) is translated to
libdatachannel ICE servers (stun/turn/turns URLs, udp/tcp/tls relays).
Candidates are emitted without their `a=` prefix, the form browsers and
the Swift SDK expect; incoming candidates are accepted either way.
`restart?` on create-offer is accepted and ignored: the room only asks
for a restart on a fresh connection.
`node-datachannel` keeps the process alive while it is initialized, so
the factory counts live connections and calls its `cleanup()` once the
last one closes; the next connection re-initializes the library.
`node-datachannel` is an optionalDependency: nothing in this namespace
loads it until the first connection is made, and a missing module
surfaces as a :ird.p2p/config-error from `make-peer-connection`.
Internal interop namespace: js/require, ^js property access and JS
callbacks live only here."
(:require [clojure.string :as str]
[cljs.core.async :as async :refer [go]]
[ird.p2p.async :as a]
[ird.p2p.errors :as err]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.node :as node]
[ird.p2p.webrtc :as webrtc]))
(def ^:private ndc-module
(delay (try (js/require "node-datachannel") (catch :default e e))))
(defonce ^:private live-connections (atom 0))
(defn- acquire! [] (swap! live-connections inc) nil)
(defn- release! [^js ndc]
(when (zero? (swap! live-connections (fn [n] (max 0 (dec n)))))
(try (.cleanup ndc) (catch :default _ nil)))
nil)
;; ---- ICE configuration -------------------------------------------------------------------
(def ^:private ice-url-re #"^(stuns?|turns?):(?:\[([^\]]+)\]|([^:?/]+))(?::(\d+))?(?:\?transport=(udp|tcp))?$")
(defn ice-server-entries
"The libdatachannel ICE-server maps for one {:urls [...] :username :credential}
server: `{:hostname :port}` for stun, plus `:username :password :relayType`
(TurnUdp / TurnTcp / TurnTls) for turn and turns. Unparseable URLs are skipped."
[{:keys [urls username credential]}]
(keep (fn [url]
(when-let [[_ scheme host6 host port transport] (re-matches ice-url-re (str url))]
(let [turn? (str/starts-with? scheme "turn")
tls? (= "turns" scheme)
hostname (or host6 host)
port (if port (js/parseInt port 10) (if tls? 5349 3478))]
(cond-> {:hostname hostname :port port}
turn? (assoc :username (or username "") :password (or credential "")
:relayType (cond tls? "TurnTls" (= "tcp" transport) "TurnTcp" :else "TurnUdp"))))))
urls))
(defn- rtc-config [{:keys [servers relay-only?]}]
(clj->js {:iceServers (vec (mapcat ice-server-entries servers))
:iceTransportPolicy (if relay-only? "relay" "all")
:disableAutoNegotiation true}))
;; ---- the ^js interop seam --------------------------------------------------------------------
(defn- new-pc [^js ndc name cfg] (let [C (.-PeerConnection ndc)] (C. name cfg)))
(defn- new-dc [^js pc] (.createDataChannel pc "ird" #js {:negotiated true :id 0}))
(defn- pc-set-local! [^js pc type] (.setLocalDescription pc type))
(defn- pc-set-remote! [^js pc sdp type] (.setRemoteDescription pc sdp type))
(defn- pc-local-sdp [^js pc] (some-> (.localDescription pc) .-sdp))
(defn- pc-add-candidate! [^js pc candidate mid] (.addRemoteCandidate pc candidate mid))
(defn- pc-close! [^js pc] (.close pc))
(defn- pc-on-local-candidate [^js pc f] (.onLocalCandidate pc f))
(defn- pc-on-gathering-state [^js pc f] (.onGatheringStateChange pc f))
(defn- pc-on-state [^js pc f] (.onStateChange pc f))
(defn- dc-open? [^js dc] (.isOpen dc))
(defn- dc-send-text! [^js dc text] (.sendMessage dc text))
(defn- dc-send-binary! [^js dc buf] (.sendMessageBinary dc buf))
(defn- dc-close! [^js dc] (.close dc))
(defn- dc-on-open [^js dc f] (.onOpen dc f))
(defn- dc-on-closed [^js dc f] (.onClosed dc f))
(defn- dc-on-error [^js dc f] (.onError dc f))
(defn- dc-on-message [^js dc f] (.onMessage dc f))
(defn- dc-on-buffered-low [^js dc f] (.onBufferedAmountLow dc f))
(defn- candidate-string [c] (str/replace-first (str c) #"^a=" ""))
(defn- message->event [m]
(if (string? m)
{:type :message :data m :binary? false}
{:type :message :data (bytes/buffer->b64 m) :binary? true}))
(defn- failure [what e]
(err/transport-error (str what ": " (node/error-message e)) {}))
;; ---- the connection ----------------------------------------------------------------------------
(defn- make-connection [ndc ice]
(let [pc (new-pc ndc "ird" (rtc-config ice))
dc (new-dc pc)
{:keys [out put! close!]} (a/unbounded-chan)
st (atom {:open? false :closed? false})
emit! (fn [ev] (when-not (:closed? @st) (put! ev)))]
(acquire!)
(pc-on-local-candidate pc (fn [c mid]
(emit! {:type :ice-candidate :candidate (candidate-string c)
:sdp-mid (when (string? mid) mid) :sdp-mline-index nil})))
(pc-on-gathering-state pc (fn [s] (when (= "complete" s) (emit! {:type :ice-gathering-complete}))))
(pc-on-state pc (fn [s] (emit! {:type :connection-state :state (keyword (str s))})))
(dc-on-open dc (fn [] (swap! st assoc :open? true) (emit! {:type :data-channel-open})))
(dc-on-closed dc (fn [] (when (:open? @st) (swap! st assoc :open? false) (emit! {:type :data-channel-closed}))))
(dc-on-error dc (fn [_] (when (:open? @st) (swap! st assoc :open? false) (emit! {:type :data-channel-closed}))))
(dc-on-buffered-low dc (fn [] (emit! {:type :buffered-amount-low})))
(dc-on-message dc (fn [m] (emit! (message->event m))))
(reify webrtc/PeerConnection
(events [_] out)
(create-offer [_ _restart?]
(go (try (pc-set-local! pc "offer")
(or (pc-local-sdp pc) (err/transport-error "createOffer produced no local description" {}))
(catch :default e (failure "createOffer failed" e)))))
(create-answer [_]
(go (try (pc-set-local! pc "answer")
(or (pc-local-sdp pc) (err/transport-error "createAnswer produced no local description" {}))
(catch :default e (failure "createAnswer failed" e)))))
(set-remote-description [_ type sdp]
(go (try (pc-set-remote! pc (str sdp) (name type)) true
(catch :default e (failure "setRemoteDescription failed" e)))))
(add-ice-candidate [_ candidate sdp-mid _sdp-mline-index]
(go (try (pc-add-candidate! pc (str candidate) (or sdp-mid "0")) true
(catch :default e (failure "addIceCandidate failed" e)))))
(send-text [_ text]
(go (try (if (and (:open? @st) (dc-open? dc) (dc-send-text! dc text))
true
(err/transport-error "data channel not open" {}))
(catch :default e (failure "send failed" e)))))
(send-binary [_ data-b64]
(go (try (if (and (:open? @st) (dc-open? dc) (dc-send-binary! dc (bytes/b64->buffer data-b64)))
true
(err/transport-error "data channel not open" {}))
(catch :default e (failure "send failed" e)))))
(close [_]
(go
(when-not (:closed? @st)
(let [was-open (:open? @st)]
(swap! st assoc :open? false)
(when was-open (put! {:type :data-channel-closed}))
(put! {:type :connection-state :state :closed})
(swap! st assoc :closed? true)
(try (dc-close! dc) (catch :default _ nil))
(try (pc-close! pc) (catch :default _ nil))
(close!)
(release! ndc)))
nil)))))
(defn wrtc-factory
"A PeerConnectionFactory over node-datachannel. Each
`make-peer-connection` delivers a fresh connection, or a config error
when `node-datachannel` is not installed (it is an optionalDependency)."
[]
(reify webrtc/PeerConnectionFactory
(make-peer-connection [_ ice]
(go
(let [ndc @ndc-module]
(if (node/error? ndc)
(err/config-error (str "node-datachannel is not available: " (node/error-message ndc)
" — install it (npm install node-datachannel) to use ird.p2p.impl.wrtc")
{})
(try (make-connection ndc ice)
(catch :default e (failure "PeerConnection failed" e)))))))))
|