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 | (ns ird.p2p.webrtc
"The peer-connection SPI rooms negotiate through, and an in-memory
factory that pairs connections inside one process for tests.
Real WebRTC on Node: pass ird.p2p.impl.wrtc/wrtc-factory (over the
optional `node-datachannel` module) as :pc-factory to ird.p2p/make-app,
or implement `PeerConnectionFactory`/`PeerConnection` over any other
library (`werift`, ...). Each connection owns one negotiated data channel
(id 0, label \"ird\", ordered, reliable) created on both sides before any
description is exchanged. Events are plain maps on the connection's
channel:
{:type :ice-candidate :candidate <string> :sdp-mid <string?> :sdp-mline-index <int?>}
{:type :ice-gathering-complete}
{:type :connection-state :state :new/:connecting/:connected/:disconnected/:failed/:closed}
{:type :data-channel-open} {:type :data-channel-closed}
{:type :message :data <string or b64> :binary? <bool>}
{:type :buffered-amount-low}"
(:require [cljs.core.async :as async :refer [go <!]]
[ird.p2p.async :as a]
[ird.p2p.errors :as err]
[ird.p2p.impl.bytes :as bytes]))
(defprotocol PeerConnectionFactory
(make-peer-connection [factory ice]
"A new connection with ICE configuration {:servers [...] :relay-only? bool}.
Channel of a PeerConnection or an error."))
(defprotocol PeerConnection
(events [pc] "The channel of event maps; closes when the connection closes.")
(create-offer [pc restart?] "Create and apply a local offer. Channel of its SDP or an error.")
(create-answer [pc] "Create and apply a local answer to the remote offer. Channel of its SDP.")
(set-remote-description [pc type sdp] "TYPE is :offer or :answer. Channel of true or an error.")
(add-ice-candidate [pc candidate sdp-mid sdp-mline-index] "Channel of true or an error.")
(send-text [pc text] "Channel of true, or an error when the channel is not open.")
(send-binary [pc data-b64] "Channel of true, or an error when the channel is not open.")
(close [pc] "Channel that closes once torn down."))
;; ---- in-memory factory ---------------------------------------------------------------
(defn- sdp [type id restart?]
(str "v=0\r\ns=inmemory\r\na=type:" (name type) "\r\na=handle:" id "\r\n" (when restart? "a=restart\r\n")))
(defn- handle-id [sdp-text]
(second (re-find #"a=handle:([^\r\n]+)" (str sdp-text))))
(declare in-memory-connection)
(defn in-memory-factory
"Pairs peer connections inside one process. Offers and answers carry the
handle's id, so two handles that exchange them through the room's
signaling end up connected exactly like real WebRTC peers would — with no
network. Returns the factory; `fail-all!`, `fail-next!` and
`connection-attempts` drive failure scenarios in tests."
[]
(let [network (atom {:handles {} :pairs {} :failures-left 0 :attempts 0})]
(with-meta
(reify PeerConnectionFactory
(make-peer-connection [_ _ice]
(go (in-memory-connection network))))
{:network network})))
(defn- network-of [factory] (:network (meta factory)))
(defn fail-all!
"Break every established pair as if the network path died."
[factory]
(let [network (network-of factory)
pairs (:pairs @network)]
(swap! network assoc :pairs {})
(doseq [[id _] pairs]
(when-let [h (get-in @network [:handles id])] ((:fail! h))))
nil))
(defn fail-next!
"Make the next N connection attempts fail after negotiation."
[factory n]
(swap! (network-of factory) update :failures-left + n)
nil)
(defn connection-attempts
"How many negotiations reached the connect step (successful or failed)."
[factory]
(:attempts @(network-of factory)))
(defn- network-connect! [network a b]
(let [{:keys [handles failures-left]} @network
ha (get handles a) hb (get handles b)]
(when (and ha hb)
(swap! network update :attempts inc)
(if (pos? failures-left)
(do (swap! network update :failures-left dec)
((:fail! ha)) ((:fail! hb)))
(do (swap! network update :pairs assoc a b b a)
((:connected! ha)) ((:connected! hb)))))))
(defn- network-unregister! [network id]
(let [other (get-in @network [:pairs id])]
(swap! network (fn [n] (-> n (update :handles dissoc id) (update :pairs dissoc id other))))
(when other
(when-let [h (get-in @network [:handles other])] ((:peer-closed! h))))))
(defn- network-deliver! [network from data binary?]
(when-let [to (get-in @network [:pairs from])]
(when-let [h (get-in @network [:handles to])]
((:receive! h) data binary?))))
(defn- in-memory-connection [network]
(let [id (bytes/random-hex 8)
{:keys [out put! close!]} (a/unbounded-chan)
st (atom {:state :stable :remote nil :open? false :closed? false})
pc (reify PeerConnection
(events [_] out)
(create-offer [_ restart?]
(go
(swap! st assoc :state :have-local-offer)
(put! {:type :ice-candidate :candidate "candidate:1 1 udp 1 127.0.0.1 9 typ host" :sdp-mid "0" :sdp-mline-index 0})
(put! {:type :ice-gathering-complete})
(sdp :offer id restart?)))
(create-answer [_]
(go
(let [remote (:remote @st)]
(swap! st assoc :state :stable)
(if (nil? remote)
(err/config-error "createAnswer without a remote offer" {})
(do
(put! {:type :ice-candidate :candidate "candidate:1 1 udp 1 127.0.0.1 9 typ host" :sdp-mid "0" :sdp-mline-index 0})
(put! {:type :ice-gathering-complete})
(put! {:type :connection-state :state :connecting})
(network-connect! network remote id)
(sdp :answer id false))))))
(set-remote-description [_ type sdp-text]
(go
(if-let [other (handle-id sdp-text)]
(do (swap! st assoc :remote other :state (if (= :offer type) :have-remote-offer :stable))
(when (= :answer type) (put! {:type :connection-state :state :connecting}))
true)
(err/config-error "bad in-memory SDP" {}))))
(add-ice-candidate [_ _ _ _] (go true))
(send-text [_ text]
(go (if (:open? @st)
(do (network-deliver! network id text false) true)
(err/transport-error "data channel not open" {}))))
(send-binary [_ data-b64]
(go (if (:open? @st)
(do (network-deliver! network id data-b64 true) true)
(err/transport-error "data channel not open" {}))))
(close [_]
(go
(let [was-open (:open? @st)]
(swap! st assoc :closed? true :open? false)
(network-unregister! network id)
(when was-open (put! {:type :data-channel-closed}))
(put! {:type :connection-state :state :closed})
(close!)
nil))))]
(swap! network assoc-in [:handles id]
{:connected! (fn [] (swap! st assoc :open? true)
(put! {:type :connection-state :state :connected})
(put! {:type :data-channel-open}))
:fail! (fn [] (let [was-open (:open? @st)]
(swap! st assoc :open? false)
(when was-open (put! {:type :data-channel-closed}))
(put! {:type :connection-state :state :failed})))
:peer-closed! (fn [] (let [was-open (:open? @st)]
(swap! st assoc :open? false)
(when was-open (put! {:type :data-channel-closed}))
(put! {:type :connection-state :state :disconnected})))
:receive! (fn [data binary?] (put! {:type :message :data data :binary? binary?}))})
pc))
|