February 2016 - tor-commits - lists.torproject.org

[snowflake/master] another goconvey test for RequestOffer, fix data race in the Broker test
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit 661286894a2c3a20ef0dba207314291c14c24708 Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 12:52:22 2016 -0800 another goconvey test for RequestOffer, fix data race in the Broker test --- broker/snowflake-broker_test.go | 31 +++++++++++++++++++++++-------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/broker/snowflake-broker_test.go b/broker/snowflake-broker_test.go index 0e125cc..44940a3 100644 --- a/broker/snowflake-broker_test.go +++ b/broker/snowflake-broker_test.go @@ -22,20 +22,35 @@ func TestBroker(t *testing.T) { So(len(ctx.idToSnowflake), ShouldEqual, 1) }) - /* Convey("Broker goroutine matches clients with proxies", func() { - ctx2 := NewBrokerContext() p := new(ProxyPoll) p.id = "test" + p.offerChannel = make(chan []byte) + go func(ctx *BrokerContext) { + ctx.proxyPolls <- p + close(ctx.proxyPolls) + }(ctx) + ctx.Broker() + So(ctx.snowflakes.Len(), ShouldEqual, 1) + snowflake := heap.Pop(ctx.snowflakes).(*Snowflake) + snowflake.offerChannel <- []byte("test offer") + offer := <-p.offerChannel + So(ctx.idToSnowflake["test"], ShouldNotBeNil) + So(offer, ShouldResemble, []byte("test offer")) + So(ctx.snowflakes.Len(), ShouldEqual, 0) + }) + + Convey("Request an offer from the Snowflake Heap", func() { + done := make(chan []byte) go func() { - ctx2.proxyPolls <- p - close(ctx2.proxyPolls) + offer := ctx.RequestOffer("test") + done <- offer }() - ctx2.Broker() - So(ctx2.snowflakes.Len(), ShouldEqual, 1) - So(ctx2.idToSnowflake["test"], ShouldNotBeNil) + request := <-ctx.proxyPolls + request.offerChannel <- []byte("test offer") + offer := <-done + So(offer, ShouldResemble, []byte("test offer")) }) - */ Convey("Responds to client offers...", func() { w := httptest.NewRecorder()

1 0

[snowflake/master] Simplify proxy poll handler, and broker match test
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit 791f6925ec749a28ad95c76325f802bc4de2d75c Author: Serene Han <keroserene+git(a)gmail.com> Date: Tue Feb 16 20:50:00 2016 -0800 Simplify proxy poll handler, and broker match test --- broker/broker.go | 89 ++++++++++++++++++++++------------------- broker/snowflake-broker_test.go | 25 +++++++++--- broker/snowflake-heap.go | 4 ++ 3 files changed, 70 insertions(+), 48 deletions(-) diff --git a/broker/broker.go b/broker/broker.go index 69b8369..9e5ee30 100644 --- a/broker/broker.go +++ b/broker/broker.go @@ -28,17 +28,17 @@ type BrokerContext struct { snowflakes *SnowflakeHeap // Map keeping track of snowflakeIDs required to match SDP answers from // the second http POST. - snowflakeMap map[string]*Snowflake - createChannel chan *ProxyRequest + snowflakeMap map[string]*Snowflake + proxyPolls chan *ProxyPoll } func NewBrokerContext() *BrokerContext { snowflakes := new(SnowflakeHeap) heap.Init(snowflakes) return &BrokerContext{ - snowflakes: snowflakes, - snowflakeMap: make(map[string]*Snowflake), - createChannel: make(chan *ProxyRequest), + snowflakes: snowflakes, + snowflakeMap: make(map[string]*Snowflake), + proxyPolls: make(chan *ProxyPoll), } } @@ -51,46 +51,58 @@ func (sh SnowflakeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { sh.h(sh.BrokerContext, w, r) } -type ProxyRequest struct { - id string - offerChan chan []byte +// Proxies may poll for client offers concurrently. +type ProxyPoll struct { + id string + offerChannel chan []byte } -// Create and add a Snowflake to the heap. -func (sc *BrokerContext) AddSnowflake(id string) *Snowflake { - snowflake := new(Snowflake) - snowflake.id = id - snowflake.clients = 0 - snowflake.offerChannel = make(chan []byte) - snowflake.answerChannel = make(chan []byte) - heap.Push(sc.snowflakes, snowflake) - sc.snowflakeMap[id] = snowflake - return snowflake +// Registers a Snowflake and waits for some Client to send an offer, +// as part of the polling logic of the proxy handler. +func (ctx *BrokerContext) RequestOffer(id string) []byte { + request := new(ProxyPoll) + request.id = id + request.offerChannel = make(chan []byte) + ctx.proxyPolls <- request + // Block until an offer is available... + offer := <-request.offerChannel + return offer } -// Match proxies to clients. -// func (ctx *BrokerContext) Broker(proxies <-chan *ProxyRequest) { +// goroutine which match proxies to clients. +// Safely processes proxy requests, responding to them with either an available +// client offer or nil on timeout / none are available. func (ctx *BrokerContext) Broker() { - // for p := range proxies { - for p := range ctx.createChannel { - snowflake := ctx.AddSnowflake(p.id) - // Wait for a client to avail an offer to the snowflake, or timeout - // and ask the snowflake to poll later. - go func(p *ProxyRequest) { + for request := range ctx.proxyPolls { + snowflake := ctx.AddSnowflake(request.id) + // Wait for a client to avail an offer to the snowflake. + go func(request *ProxyPoll) { select { case offer := <-snowflake.offerChannel: - log.Println("Passing client offer to snowflake.") - p.offerChan <- offer + log.Println("Passing client offer to snowflake proxy.") + request.offerChannel <- offer case <-time.After(time.Second * ProxyTimeout): // This snowflake is no longer available to serve clients. heap.Remove(ctx.snowflakes, snowflake.index) delete(ctx.snowflakeMap, snowflake.id) - p.offerChan <- nil + request.offerChannel <- nil } - }(p) + }(request) } } +// Create and add a Snowflake to the heap. +func (ctx *BrokerContext) AddSnowflake(id string) *Snowflake { + snowflake := new(Snowflake) + snowflake.id = id + snowflake.clients = 0 + snowflake.offerChannel = make(chan []byte) + snowflake.answerChannel = make(chan []byte) + heap.Push(ctx.snowflakes, snowflake) + ctx.snowflakeMap[id] = snowflake + return snowflake +} + func robotsTxtHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/plain; charset=utf-8") w.Write([]byte("User-agent: *\nDisallow:\n")) @@ -145,14 +157,15 @@ func clientHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { case answer := <-snowflake.answerChannel: log.Println("Client: Retrieving answer") w.Write(answer) - // Only remove from the snowflake map once the answer is set. - delete(ctx.snowflakeMap, snowflake.id) case <-time.After(time.Second * ClientTimeout): log.Println("Client: Timed out.") w.WriteHeader(http.StatusGatewayTimeout) w.Write([]byte("timed out waiting for answer!")) } + // Remove from the snowflake map whether answer was sent or not, because + // this client request is now over. + delete(ctx.snowflakeMap, snowflake.id) } /* @@ -172,17 +185,9 @@ func proxyHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { if string(body) != id { // Mismatched IDs! w.WriteHeader(http.StatusBadRequest) } - // Maybe confirm that X-Session-ID is the same. log.Println("Received snowflake: ", id) - - p := new(ProxyRequest) - p.id = id - p.offerChan = make(chan []byte) - ctx.createChannel <- p - - // Wait for a client to avail an offer to the snowflake, or timeout - // and ask the snowflake to poll later. - offer := <-p.offerChan + // Wait for a client to avail an offer to the snowflake, or timeout if nil. + offer := ctx.RequestOffer(id) if nil == offer { log.Println("Proxy " + id + " did not receive a Client offer.") w.WriteHeader(http.StatusGatewayTimeout) diff --git a/broker/snowflake-broker_test.go b/broker/snowflake-broker_test.go index ee984b0..b9432d8 100644 --- a/broker/snowflake-broker_test.go +++ b/broker/snowflake-broker_test.go @@ -22,6 +22,19 @@ func TestBroker(t *testing.T) { So(len(ctx.snowflakeMap), ShouldEqual, 1) }) + Convey("Broker goroutine matches clients with proxies", func() { + p := new(ProxyPoll) + p.id = "test" + p.offerChannel = make(chan []byte) + go func() { + ctx.proxyPolls <- p + close(ctx.proxyPolls) + }() + ctx.Broker() + So(ctx.snowflakes.Len(), ShouldEqual, 1) + So(ctx.snowflakes.Len(), ShouldEqual, 1) + }) + Convey("Responds to client offers...", func() { w := httptest.NewRecorder() data := bytes.NewReader([]byte("test")) @@ -83,9 +96,9 @@ func TestBroker(t *testing.T) { done <- true }(ctx) // Pass a fake client offer to this proxy - p := <-ctx.createChannel + p := <-ctx.proxyPolls So(p.id, ShouldEqual, "test") - p.offerChan <- []byte("fake offer") + p.offerChannel <- []byte("fake offer") <-done So(w.Code, ShouldEqual, http.StatusOK) So(w.Body.String(), ShouldEqual, "fake offer") @@ -96,10 +109,10 @@ func TestBroker(t *testing.T) { proxyHandler(ctx, w, r) done <- true }(ctx) - p := <-ctx.createChannel + p := <-ctx.proxyPolls So(p.id, ShouldEqual, "test") // nil means timeout - p.offerChan <- nil + p.offerChannel <- nil <-done So(w.Body.String(), ShouldEqual, "") So(w.Code, ShouldEqual, http.StatusGatewayTimeout) @@ -159,12 +172,12 @@ func TestBroker(t *testing.T) { }() // Manually do the Broker goroutine action here for full control. - p := <-ctx.createChannel + p := <-ctx.proxyPolls So(p.id, ShouldEqual, "test") s := ctx.AddSnowflake(p.id) go func() { offer := <-s.offerChannel - p.offerChan <- offer + p.offerChannel <- offer }() So(ctx.snowflakeMap["test"], ShouldNotBeNil) diff --git a/broker/snowflake-heap.go b/broker/snowflake-heap.go index d37228f..cf249fe 100644 --- a/broker/snowflake-heap.go +++ b/broker/snowflake-heap.go @@ -4,6 +4,10 @@ Keeping track of pending available snowflake proxies. package snowflake_broker +/* +The Snowflake struct contains a single interaction +over the offer and answer channels. +*/ type Snowflake struct { id string offerChannel chan []byte

1 0

[snowflake/master] move CORS early return into the ServeHTTP wrapper, rename handlers
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit 2ae6559001ab779a0c8e9b4b6a6804353b19222a Author: Serene Han <keroserene+git(a)gmail.com> Date: Tue Feb 16 21:00:30 2016 -0800 move CORS early return into the ServeHTTP wrapper, rename handlers --- broker/broker.go | 140 +++++++++++++++++++--------------------- broker/snowflake-broker_test.go | 47 +++++++------- 2 files changed, 89 insertions(+), 98 deletions(-) diff --git a/broker/broker.go b/broker/broker.go index 9e5ee30..21777bd 100644 --- a/broker/broker.go +++ b/broker/broker.go @@ -28,27 +28,34 @@ type BrokerContext struct { snowflakes *SnowflakeHeap // Map keeping track of snowflakeIDs required to match SDP answers from // the second http POST. - snowflakeMap map[string]*Snowflake - proxyPolls chan *ProxyPoll + idToSnowflake map[string]*Snowflake + proxyPolls chan *ProxyPoll } func NewBrokerContext() *BrokerContext { snowflakes := new(SnowflakeHeap) heap.Init(snowflakes) return &BrokerContext{ - snowflakes: snowflakes, - snowflakeMap: make(map[string]*Snowflake), - proxyPolls: make(chan *ProxyPoll), + snowflakes: snowflakes, + idToSnowflake: make(map[string]*Snowflake), + proxyPolls: make(chan *ProxyPoll), } } +// Implements the http.Handler interface type SnowflakeHandler struct { *BrokerContext - h func(*BrokerContext, http.ResponseWriter, *http.Request) + handle func(*BrokerContext, http.ResponseWriter, *http.Request) } func (sh SnowflakeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - sh.h(sh.BrokerContext, w, r) + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Headers", "Origin, X-Session-ID") + // Return early if it's CORS preflight. + if "OPTIONS" == r.Method { + return + } + sh.handle(sh.BrokerContext, w, r) } // Proxies may poll for client offers concurrently. @@ -69,12 +76,14 @@ func (ctx *BrokerContext) RequestOffer(id string) []byte { return offer } -// goroutine which match proxies to clients. +// goroutine which matches clients to proxies and sends SDP offers along. // Safely processes proxy requests, responding to them with either an available // client offer or nil on timeout / none are available. func (ctx *BrokerContext) Broker() { for request := range ctx.proxyPolls { snowflake := ctx.AddSnowflake(request.id) + // defer heap.Remove(ctx.snowflakes, snowflake.index) + // defer delete(ctx.idToSnowflake, snowflake.id) // Wait for a client to avail an offer to the snowflake. go func(request *ProxyPoll) { select { @@ -83,8 +92,9 @@ func (ctx *BrokerContext) Broker() { request.offerChannel <- offer case <-time.After(time.Second * ProxyTimeout): // This snowflake is no longer available to serve clients. + // TODO: Fix race using a delete channel heap.Remove(ctx.snowflakes, snowflake.index) - delete(ctx.snowflakeMap, snowflake.id) + delete(ctx.idToSnowflake, snowflake.id) request.offerChannel <- nil } }(request) @@ -99,32 +109,34 @@ func (ctx *BrokerContext) AddSnowflake(id string) *Snowflake { snowflake.offerChannel = make(chan []byte) snowflake.answerChannel = make(chan []byte) heap.Push(ctx.snowflakes, snowflake) - ctx.snowflakeMap[id] = snowflake + ctx.idToSnowflake[id] = snowflake return snowflake } -func robotsTxtHandler(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "text/plain; charset=utf-8") - w.Write([]byte("User-agent: *\nDisallow:\n")) -} - -func ipHandler(w http.ResponseWriter, r *http.Request) { - remoteAddr := r.RemoteAddr - if net.ParseIP(remoteAddr).To4() == nil { - remoteAddr = "[" + remoteAddr + "]" +/* +For snowflake proxies to request a client from the Broker. +*/ +func proxyPolls(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { + id := r.Header.Get("X-Session-ID") + body, err := ioutil.ReadAll(r.Body) + if nil != err { + log.Println("Invalid data.") + w.WriteHeader(http.StatusBadRequest) + return } - w.Header().Set("Content-Type", "text/plain; charset=utf-8") - w.Write([]byte(remoteAddr)) -} - -// Return early if it's CORS preflight. -func isPreflight(w http.ResponseWriter, r *http.Request) bool { - w.Header().Set("Access-Control-Allow-Origin", "*") - w.Header().Set("Access-Control-Allow-Headers", "Origin, X-Session-ID") - if "OPTIONS" == r.Method { - return true + if string(body) != id { // Mismatched IDs! + w.WriteHeader(http.StatusBadRequest) } - return false + log.Println("Received snowflake: ", id) + // Wait for a client to avail an offer to the snowflake, or timeout if nil. + offer := ctx.RequestOffer(id) + if nil == offer { + log.Println("Proxy " + id + " did not receive a Client offer.") + w.WriteHeader(http.StatusGatewayTimeout) + return + } + log.Println("Passing client offer to snowflake.") + w.Write(offer) } /* @@ -132,15 +144,13 @@ Expects a WebRTC SDP offer in the Request to give to an assigned snowflake proxy, which responds with the SDP answer to be sent in the HTTP response back to the client. */ -func clientHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { +func clientOffers(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { offer, err := ioutil.ReadAll(r.Body) if nil != err { log.Println("Invalid data.") w.WriteHeader(http.StatusBadRequest) return } - w.Header().Set("Access-Control-Allow-Origin", "*") - w.Header().Set("Access-Control-Allow-Headers", "X-Session-ID") // Find the most available snowflake proxy, and pass the offer to it. // TODO: Needs improvement - maybe shouldn't immediately fail? @@ -150,6 +160,7 @@ func clientHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { return } snowflake := heap.Pop(ctx.snowflakes).(*Snowflake) + defer delete(ctx.idToSnowflake, snowflake.id) snowflake.offerChannel <- offer // Wait for the answer to be returned on the channel. @@ -163,38 +174,6 @@ func clientHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusGatewayTimeout) w.Write([]byte("timed out waiting for answer!")) } - // Remove from the snowflake map whether answer was sent or not, because - // this client request is now over. - delete(ctx.snowflakeMap, snowflake.id) -} - -/* -For snowflake proxies to request a client from the Broker. -*/ -func proxyHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { - if isPreflight(w, r) { - return - } - id := r.Header.Get("X-Session-ID") - body, err := ioutil.ReadAll(r.Body) - if nil != err { - log.Println("Invalid data.") - w.WriteHeader(http.StatusBadRequest) - return - } - if string(body) != id { // Mismatched IDs! - w.WriteHeader(http.StatusBadRequest) - } - log.Println("Received snowflake: ", id) - // Wait for a client to avail an offer to the snowflake, or timeout if nil. - offer := ctx.RequestOffer(id) - if nil == offer { - log.Println("Proxy " + id + " did not receive a Client offer.") - w.WriteHeader(http.StatusGatewayTimeout) - return - } - log.Println("Passing client offer to snowflake.") - w.Write(offer) } /* @@ -202,15 +181,12 @@ Expects snowflake proxes which have previously successfully received an offer from proxyHandler to respond with an answer in an HTTP POST, which the broker will pass back to the original client. */ -func answerHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { - if isPreflight(w, r) { - return - } +func proxyAnswers(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { id := r.Header.Get("X-Session-ID") - snowflake, ok := ctx.snowflakeMap[id] + snowflake, ok := ctx.idToSnowflake[id] if !ok || nil == snowflake { - // The snowflake took too long to respond with an answer, - // and the designated client is no longer around / recognized by the Broker. + // The snowflake took too long to respond with an answer, so its client + // disappeared / the snowflake is no longer recognized by the Broker. w.WriteHeader(http.StatusGone) return } @@ -229,6 +205,20 @@ func debugHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { w.Write([]byte(s)) } +func robotsTxtHandler(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.Write([]byte("User-agent: *\nDisallow:\n")) +} + +func ipHandler(w http.ResponseWriter, r *http.Request) { + remoteAddr := r.RemoteAddr + if net.ParseIP(remoteAddr).To4() == nil { + remoteAddr = "[" + remoteAddr + "]" + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.Write([]byte(remoteAddr)) +} + func init() { ctx := NewBrokerContext() @@ -237,8 +227,8 @@ func init() { http.HandleFunc("/robots.txt", robotsTxtHandler) http.HandleFunc("/ip", ipHandler) - http.Handle("/client", SnowflakeHandler{ctx, clientHandler}) - http.Handle("/proxy", SnowflakeHandler{ctx, proxyHandler}) - http.Handle("/answer", SnowflakeHandler{ctx, answerHandler}) + http.Handle("/proxy", SnowflakeHandler{ctx, proxyPolls}) + http.Handle("/client", SnowflakeHandler{ctx, clientOffers}) + http.Handle("/answer", SnowflakeHandler{ctx, proxyAnswers}) http.Handle("/debug", SnowflakeHandler{ctx, debugHandler}) } diff --git a/broker/snowflake-broker_test.go b/broker/snowflake-broker_test.go index b9432d8..0e125cc 100644 --- a/broker/snowflake-broker_test.go +++ b/broker/snowflake-broker_test.go @@ -16,24 +16,26 @@ func TestBroker(t *testing.T) { Convey("Adds Snowflake", func() { So(ctx.snowflakes.Len(), ShouldEqual, 0) - So(len(ctx.snowflakeMap), ShouldEqual, 0) + So(len(ctx.idToSnowflake), ShouldEqual, 0) ctx.AddSnowflake("foo") So(ctx.snowflakes.Len(), ShouldEqual, 1) - So(len(ctx.snowflakeMap), ShouldEqual, 1) + So(len(ctx.idToSnowflake), ShouldEqual, 1) }) + /* Convey("Broker goroutine matches clients with proxies", func() { + ctx2 := NewBrokerContext() p := new(ProxyPoll) p.id = "test" - p.offerChannel = make(chan []byte) go func() { - ctx.proxyPolls <- p - close(ctx.proxyPolls) + ctx2.proxyPolls <- p + close(ctx2.proxyPolls) }() - ctx.Broker() - So(ctx.snowflakes.Len(), ShouldEqual, 1) - So(ctx.snowflakes.Len(), ShouldEqual, 1) + ctx2.Broker() + So(ctx2.snowflakes.Len(), ShouldEqual, 1) + So(ctx2.idToSnowflake["test"], ShouldNotBeNil) }) + */ Convey("Responds to client offers...", func() { w := httptest.NewRecorder() @@ -42,9 +44,7 @@ func TestBroker(t *testing.T) { So(err, ShouldBeNil) Convey("with 503 when no snowflakes are available.", func() { - clientHandler(ctx, w, r) - h := w.Header() - So(h["Access-Control-Allow-Headers"], ShouldNotBeNil) + clientOffers(ctx, w, r) So(w.Code, ShouldEqual, http.StatusServiceUnavailable) So(w.Body.String(), ShouldEqual, "") }) @@ -54,7 +54,7 @@ func TestBroker(t *testing.T) { // Prepare a fake proxy to respond with. snowflake := ctx.AddSnowflake("fake") go func() { - clientHandler(ctx, w, r) + clientOffers(ctx, w, r) done <- true }() offer := <-snowflake.offerChannel @@ -72,7 +72,8 @@ func TestBroker(t *testing.T) { done := make(chan bool) snowflake := ctx.AddSnowflake("fake") go func() { - clientHandler(ctx, w, r) + clientOffers(ctx, w, r) + // Takes a few seconds here... done <- true }() offer := <-snowflake.offerChannel @@ -92,7 +93,7 @@ func TestBroker(t *testing.T) { Convey("with a client offer if available.", func() { go func(ctx *BrokerContext) { - proxyHandler(ctx, w, r) + proxyPolls(ctx, w, r) done <- true }(ctx) // Pass a fake client offer to this proxy @@ -106,7 +107,7 @@ func TestBroker(t *testing.T) { Convey("times out when no client offer is available.", func() { go func(ctx *BrokerContext) { - proxyHandler(ctx, w, r) + proxyPolls(ctx, w, r) done <- true }(ctx) p := <-ctx.proxyPolls @@ -129,7 +130,7 @@ func TestBroker(t *testing.T) { So(err, ShouldBeNil) r.Header.Set("X-Session-ID", "test") go func(ctx *BrokerContext) { - answerHandler(ctx, w, r) + proxyAnswers(ctx, w, r) }(ctx) answer := <-s.answerChannel So(w.Code, ShouldEqual, http.StatusOK) @@ -140,7 +141,7 @@ func TestBroker(t *testing.T) { r, err := http.NewRequest("POST", "snowflake.broker/answer", nil) So(err, ShouldBeNil) r.Header.Set("X-Session-ID", "invalid") - answerHandler(ctx, w, r) + proxyAnswers(ctx, w, r) So(w.Code, ShouldEqual, http.StatusGone) }) @@ -149,7 +150,7 @@ func TestBroker(t *testing.T) { r, err := http.NewRequest("POST", "snowflake.broker/answer", data) r.Header.Set("X-Session-ID", "test") So(err, ShouldBeNil) - answerHandler(ctx, w, r) + proxyAnswers(ctx, w, r) So(w.Code, ShouldEqual, http.StatusBadRequest) }) }) @@ -167,7 +168,7 @@ func TestBroker(t *testing.T) { So(err, ShouldBeNil) rP.Header.Set("X-Session-ID", "test") go func() { - proxyHandler(ctx, wP, rP) + proxyPolls(ctx, wP, rP) polled <- true }() @@ -179,7 +180,7 @@ func TestBroker(t *testing.T) { offer := <-s.offerChannel p.offerChannel <- offer }() - So(ctx.snowflakeMap["test"], ShouldNotBeNil) + So(ctx.idToSnowflake["test"], ShouldNotBeNil) // Client request blocks until proxy answer arrives. dataC := bytes.NewReader([]byte("fake offer")) @@ -187,21 +188,21 @@ func TestBroker(t *testing.T) { rC, err := http.NewRequest("POST", "snowflake.broker/client", dataC) So(err, ShouldBeNil) go func() { - clientHandler(ctx, wC, rC) + clientOffers(ctx, wC, rC) done <- true }() <-polled So(wP.Code, ShouldEqual, http.StatusOK) So(wP.Body.String(), ShouldResemble, "fake offer") - So(ctx.snowflakeMap["test"], ShouldNotBeNil) + So(ctx.idToSnowflake["test"], ShouldNotBeNil) // Follow up with the answer request afterwards wA := httptest.NewRecorder() dataA := bytes.NewReader([]byte("fake answer")) rA, err := http.NewRequest("POST", "snowflake.broker/proxy", dataA) So(err, ShouldBeNil) rA.Header.Set("X-Session-ID", "test") - answerHandler(ctx, wA, rA) + proxyAnswers(ctx, wA, rA) So(wA.Code, ShouldEqual, http.StatusOK) <-done

1 0

[snowflake/master] prepare snowflake client for buffered datachannel writes, separate out dialWebRTC (#12)
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit 760dee8a0f3efc1a71780bfde277ae7a5a7a6d9b Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 17:39:09 2016 -0800 prepare snowflake client for buffered datachannel writes, separate out dialWebRTC (#12) --- client/snowflake.go | 196 ++++++++++++++++++++++++++++++---------------------- 1 file changed, 115 insertions(+), 81 deletions(-) diff --git a/client/snowflake.go b/client/snowflake.go index ba1561e..7c47fbb 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -28,7 +28,7 @@ var frontDomain string // ends, -1 is written. var handlerChan = make(chan int) -var signalChan = make(chan *webrtc.SessionDescription) +var answerChannel = make(chan *webrtc.SessionDescription) func copyLoop(a, b net.Conn) { var wg sync.WaitGroup @@ -46,10 +46,16 @@ func copyLoop(a, b net.Conn) { wg.Wait() } +// Implements net.Conn interface type webRTCConn struct { - pc *webrtc.PeerConnection - dc *webrtc.DataChannel - recvPipe *io.PipeReader + pc *webrtc.PeerConnection + dc *webrtc.DataChannel + broker *BrokerChannel + recvPipe *io.PipeReader + writePipe *io.PipeWriter + offerChannel chan *webrtc.SessionDescription + errorChannel chan error + openChannel chan struct{} } var webrtcRemote *webRTCConn @@ -61,6 +67,7 @@ func (c *webRTCConn) Read(b []byte) (int, error) { func (c *webRTCConn) Write(b []byte) (int, error) { // log.Printf("webrtc Write %d %+q", len(b), string(b)) log.Printf("Write %d bytes --> WebRTC", len(b)) + // Buffer in case datachannel isn't available. c.dc.Send(b) return len(b), nil } @@ -90,18 +97,87 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error { return fmt.Errorf("SetWriteDeadline not implemented") } -func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( - *webRTCConn, error) { +// Create a WebRTC DataChannel locally. +// This triggers "OnNegotiationNeeded" which should prepare an SDP offer. +func (c *webRTCConn) EstablishDataChannel() error { + dc, err := c.pc.CreateDataChannel("snowflake", webrtc.Init{}) + if err != nil { + log.Printf("CreateDataChannel: %s", err) + return err + } + dc.OnOpen = func() { + log.Println("OnOpen channel") + c.openChannel <- struct{}{} + } + dc.OnClose = func() { + log.Println("OnClose channel") + // writePipe.Close() + close(c.openChannel) + // TODO: (Issue #12) Should attempt to renegotiate at this point. + } + dc.OnMessage = func(msg []byte) { + log.Printf("OnMessage <--- %d bytes", len(msg)) + n, err := c.writePipe.Write(msg) + if err != nil { + // TODO: Maybe shouldn't actually close. + c.writePipe.CloseWithError(err) + } + if n != len(msg) { + panic("short write") + } + } + c.dc = dc + return nil +} - offerChan := make(chan *webrtc.SessionDescription) - errChan := make(chan error) - openChan := make(chan struct{}) +// Block until an offer is available, then send it to either +// the Broker or signal pipe. +func (c *webRTCConn) sendOffer() error { + select { + case offer := <-c.offerChannel: + if "" == brokerURL { + log.Printf("Please Copy & Paste the following to the peer:") + log.Printf("----------------") + fmt.Fprintln(logFile, "\n"+offer.Serialize()+"\n") + log.Printf("----------------") + return nil + } + // Use Broker... + go func() { + log.Println("Sending offer via BrokerChannel...\nTarget URL: ", brokerURL, + "\nFront URL: ", frontDomain) + answer, err := c.broker.Negotiate(c.pc.LocalDescription()) + if nil != err { + log.Printf("BrokerChannel signaling error: %s", err) + return + } + if nil == answer { + log.Printf("BrokerChannel: No answer received.") + return + // return errors.New("No answer received.") + } + answerChannel <- answer + }() + case err := <-c.errorChannel: + c.pc.Close() + return err + } + return nil +} +func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( + *webRTCConn, error) { pc, err := webrtc.NewPeerConnection(config) if err != nil { log.Printf("NewPeerConnection: %s", err) return nil, err } + connection := new(webRTCConn) + connection.broker = broker + connection.pc = pc + connection.offerChannel = make(chan *webrtc.SessionDescription) + connection.errorChannel = make(chan error) + connection.openChannel = make(chan struct{}) // Triggered by CreateDataChannel. pc.OnNegotiationNeeded = func() { @@ -109,12 +185,12 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( go func() { offer, err := pc.CreateOffer() if err != nil { - errChan <- err + connection.errorChannel <- err return } err = pc.SetLocalDescription(offer) if err != nil { - errChan <- err + connection.errorChannel <- err return } }() @@ -126,7 +202,7 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( // TODO: This may soon be deprecated, consider OnIceGatheringStateChange. pc.OnIceComplete = func() { log.Printf("OnIceComplete") - offerChan <- pc.LocalDescription() + connection.offerChannel <- pc.LocalDescription() } // This callback is not expected, as the Client initiates the creation // of the data channel, not the remote peer. @@ -135,62 +211,14 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( panic("OnDataChannel") } - pr, pw := io.Pipe() + // Pipes remain the same even when DataChannel gets switched. + connection.recvPipe, connection.writePipe = io.Pipe() - dc, err := pc.CreateDataChannel("test", webrtc.Init{}) - if err != nil { - log.Printf("CreateDataChannel: %s", err) - return nil, err - } - dc.OnOpen = func() { - log.Println("OnOpen channel") - openChan <- struct{}{} - } - dc.OnClose = func() { - log.Println("OnClose channel") - pw.Close() - close(openChan) - // TODO: (Issue #12) Should attempt to renegotiate at this point. - } - dc.OnMessage = func(msg []byte) { - log.Printf("OnMessage <--- %d bytes", len(msg)) - n, err := pw.Write(msg) - if err != nil { - pw.CloseWithError(err) - } - if n != len(msg) { - panic("short write") - } - } - - select { - case err := <-errChan: - pc.Close() - return nil, err - case offer := <-offerChan: - log.Printf("----------------") - fmt.Fprintln(logFile, "\n"+offer.Serialize()+"\n") - log.Printf("----------------") - go func() { - if "" != brokerURL { - log.Println("Sending offer via BrokerChannel...\nTarget URL: ", brokerURL, - "\nFront URL: ", frontDomain) - answer, err := broker.Negotiate(pc.LocalDescription()) - if nil != err { - log.Printf("BrokerChannel signaling error: %s", err) - } - if nil == answer { - log.Printf("BrokerChannel: No answer received.") - } else { - signalChan <- answer - } - } - }() - } - - log.Printf("waiting for answer") - answer, ok := <-signalChan + connection.EstablishDataChannel() + connection.sendOffer() + log.Printf("waiting for answer...") + answer, ok := <-answerChannel if !ok { pc.Close() return nil, fmt.Errorf("no answer received") @@ -205,13 +233,13 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( // Wait until data channel is open; otherwise for example sends may get // lost. // TODO: Buffering *should* work though. - _, ok = <-openChan + _, ok = <-connection.openChannel if !ok { pc.Close() return nil, fmt.Errorf("failed to open data channel") } - return &webRTCConn{pc: pc, dc: dc, recvPipe: pr}, nil + return connection, nil } func endWebRTC() { @@ -229,6 +257,7 @@ func endWebRTC() { } } +// Establish a WebRTC channel for SOCKS connections. func handler(conn *pt.SocksConn) error { handlerChan <- 1 defer func() { @@ -259,7 +288,6 @@ func handler(conn *pt.SocksConn) error { } copyLoop(conn, remote) - return nil } @@ -293,10 +321,10 @@ func readSignalingMessages(f *os.File) { log.Printf("ignoring invalid signal message %+q", msg) continue } - signalChan <- sdp + answerChannel <- sdp } - log.Printf("close signalChan") - close(signalChan) + log.Printf("close answerChannel") + close(answerChannel) if err := s.Err(); err != nil { log.Printf("signal FIFO: %s", err) } @@ -308,19 +336,25 @@ func main() { flag.StringVar(&brokerURL, "url", "", "URL of signaling broker") flag.StringVar(&frontDomain, "front", "", "front domain") flag.Parse() - logFile, err = os.OpenFile("snowflake.log", os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600) if err != nil { log.Fatal(err) } defer logFile.Close() log.SetOutput(logFile) - log.Println("starting") - - if "" == brokerURL { + log.Println("\nStarting Snowflake Client...") + + // Expect user to copy-paste if + // TODO: Maybe just get rid of copy-paste entirely. + if "" != brokerURL { + log.Println("Rendezvous using Broker at: ", brokerURL) + if "" != frontDomain { + log.Println("Domain fronting using:", frontDomain) + } + } else { log.Println("No HTTP signaling detected. Waiting for a \"signal\" pipe...") // This FIFO receives signaling messages. - err = syscall.Mkfifo("signal", 0600) + err := syscall.Mkfifo("signal", 0600) if err != nil { if err.(syscall.Errno) != syscall.EEXIST { log.Fatal(err) @@ -363,6 +397,7 @@ func main() { } } pt.CmethodsDone() + defer endWebRTC() var numHandlers int = 0 var sig os.Signal @@ -382,10 +417,9 @@ func main() { ln.Close() } - if syscall.SIGTERM == sig || syscall.SIGINT == sig { - endWebRTC() - return - } + // if syscall.SIGTERM == sig || syscall.SIGINT == sig { + // return + // } // wait for second signal or no more handlers sig = nil

1 0

[snowflake/master] Buffer writes to DataChannel, remove blocking on openChannel (#12)
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit eb7eb04ac01b7a7d2466958b35403c6729014a1f Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 18:38:40 2016 -0800 Buffer writes to DataChannel, remove blocking on openChannel (#12) --- client/client_test.go | 44 +++++++++++++++++++++++++ client/snowflake.go | 89 ++++++++++++++++++++++++++++----------------------- 2 files changed, 93 insertions(+), 40 deletions(-) diff --git a/client/client_test.go b/client/client_test.go new file mode 100644 index 0000000..7b8dad2 --- /dev/null +++ b/client/client_test.go @@ -0,0 +1,44 @@ +package main + +import ( + "bytes" + . "github.com/smartystreets/goconvey/convey" + "testing" +) + +type MockDataChannel struct { + destination bytes.Buffer +} + +func (m *MockDataChannel) Send(data []byte) { + m.destination.Write(data) +} + +func (*MockDataChannel) Close() error { + return nil +} + +func TestConnect(t *testing.T) { + Convey("Snowflake", t, func() { + + Convey("WebRTC Connection", func() { + c := new(webRTCConn) + So(c.buffer.Bytes(), ShouldEqual, nil) + + Convey("SendData buffers when datachannel is nil", func() { + c.sendData([]byte("test")) + c.snowflake = nil + So(c.buffer.Bytes(), ShouldResemble, []byte("test")) + }) + + Convey("SendData sends to datachannel when not nil", func() { + mock := new(MockDataChannel) + c.snowflake = mock + c.sendData([]byte("test")) + So(c.buffer.Bytes(), ShouldEqual, nil) + So(mock.destination.Bytes(), ShouldResemble, []byte("test")) + }) + }) + + }) +} diff --git a/client/snowflake.go b/client/snowflake.go index 7c47fbb..771e90b 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -3,6 +3,7 @@ package main import ( "bufio" + "bytes" "errors" "flag" "fmt" @@ -46,16 +47,22 @@ func copyLoop(a, b net.Conn) { wg.Wait() } +// Interface that matches both webrc.DataChannel and for testing. +type SnowflakeChannel interface { + Send([]byte) + Close() error +} + // Implements net.Conn interface type webRTCConn struct { pc *webrtc.PeerConnection - dc *webrtc.DataChannel + snowflake SnowflakeChannel // Interface holding the WebRTC DataChannel. broker *BrokerChannel - recvPipe *io.PipeReader - writePipe *io.PipeWriter offerChannel chan *webrtc.SessionDescription errorChannel chan error - openChannel chan struct{} + recvPipe *io.PipeReader + writePipe *io.PipeWriter + buffer bytes.Buffer } var webrtcRemote *webRTCConn @@ -66,9 +73,7 @@ func (c *webRTCConn) Read(b []byte) (int, error) { func (c *webRTCConn) Write(b []byte) (int, error) { // log.Printf("webrtc Write %d %+q", len(b), string(b)) - log.Printf("Write %d bytes --> WebRTC", len(b)) - // Buffer in case datachannel isn't available. - c.dc.Send(b) + c.sendData(b) return len(b), nil } @@ -98,21 +103,29 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error { } // Create a WebRTC DataChannel locally. -// This triggers "OnNegotiationNeeded" which should prepare an SDP offer. func (c *webRTCConn) EstablishDataChannel() error { dc, err := c.pc.CreateDataChannel("snowflake", webrtc.Init{}) + // Triggers "OnNegotiationNeeded" on the PeerConnection, which will prepare + // an SDP offer while other goroutines operating on this struct handle the + // signaling. Eventually fires "OnOpen". if err != nil { log.Printf("CreateDataChannel: %s", err) return err } dc.OnOpen = func() { - log.Println("OnOpen channel") - c.openChannel <- struct{}{} + log.Println("WebRTC: DataChannel.OnOpen") + // Flush the buffer, then enable datachannel. + // TODO: Make this more safe + dc.Send(c.buffer.Bytes()) + log.Println("Flushed ", c.buffer.Len(), " bytes") + c.buffer.Reset() + c.snowflake = dc } dc.OnClose = func() { - log.Println("OnClose channel") - // writePipe.Close() - close(c.openChannel) + // Disable the DataChannel as a write destination. + // Future writes will go to the buffer until a new DataChannel is available. + log.Println("WebRTC: DataChannel.OnClose") + c.snowflake = nil // TODO: (Issue #12) Should attempt to renegotiate at this point. } dc.OnMessage = func(msg []byte) { @@ -126,7 +139,6 @@ func (c *webRTCConn) EstablishDataChannel() error { panic("short write") } } - c.dc = dc return nil } @@ -153,8 +165,8 @@ func (c *webRTCConn) sendOffer() error { } if nil == answer { log.Printf("BrokerChannel: No answer received.") + // TODO: Should try again here. return - // return errors.New("No answer received.") } answerChannel <- answer }() @@ -165,6 +177,18 @@ func (c *webRTCConn) sendOffer() error { return nil } +func (c *webRTCConn) sendData(data []byte) { + // Buffer the data in case datachannel isn't available yet. + if nil == c.snowflake { + log.Printf("Buffered %d bytes --> WebRTC", len(data)) + c.buffer.Write(data) + return + } + log.Printf("Write %d bytes --> WebRTC", len(data)) + c.snowflake.Send(data) +} + +// Initialize a WebRTC Connection. func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( *webRTCConn, error) { pc, err := webrtc.NewPeerConnection(config) @@ -177,13 +201,14 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( connection.pc = pc connection.offerChannel = make(chan *webrtc.SessionDescription) connection.errorChannel = make(chan error) - connection.openChannel = make(chan struct{}) + // Pipes remain the same even when DataChannel gets switched. + connection.recvPipe, connection.writePipe = io.Pipe() - // Triggered by CreateDataChannel. pc.OnNegotiationNeeded = func() { log.Println("OnNegotiationNeeded") go func() { offer, err := pc.CreateOffer() + // TODO: Potentially timeout and retry if ICE isn't working. if err != nil { connection.errorChannel <- err return @@ -208,18 +233,17 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( // of the data channel, not the remote peer. pc.OnDataChannel = func(channel *webrtc.DataChannel) { log.Println("OnDataChannel") - panic("OnDataChannel") + panic("Unexpected OnDataChannel!") } - // Pipes remain the same even when DataChannel gets switched. - connection.recvPipe, connection.writePipe = io.Pipe() - connection.EstablishDataChannel() - connection.sendOffer() + // TODO: Make this part of a re-establishment loop. + connection.sendOffer() log.Printf("waiting for answer...") answer, ok := <-answerChannel if !ok { + // TODO: Don't just fail, try again! pc.Close() return nil, fmt.Errorf("no answer received") } @@ -230,15 +254,6 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( return nil, err } - // Wait until data channel is open; otherwise for example sends may get - // lost. - // TODO: Buffering *should* work though. - _, ok = <-connection.openChannel - if !ok { - pc.Close() - return nil, fmt.Errorf("failed to open data channel") - } - return connection, nil } @@ -247,9 +262,9 @@ func endWebRTC() { if nil == webrtcRemote { return } - if nil != webrtcRemote.dc { + if nil != webrtcRemote.snowflake { log.Printf("WebRTC: closing DataChannel") - webrtcRemote.dc.Close() + webrtcRemote.snowflake.Close() } if nil != webrtcRemote.pc { log.Printf("WebRTC: closing PeerConnection") @@ -332,7 +347,7 @@ func readSignalingMessages(f *os.File) { func main() { var err error - + webrtc.SetLoggingVerbosity(1) flag.StringVar(&brokerURL, "url", "", "URL of signaling broker") flag.StringVar(&frontDomain, "front", "", "front domain") flag.Parse() @@ -368,8 +383,6 @@ func main() { go readSignalingMessages(signalFile) } - webrtc.SetLoggingVerbosity(1) - ptInfo, err = pt.ClientSetup(nil) if err != nil { log.Fatal(err) @@ -417,10 +430,6 @@ func main() { ln.Close() } - // if syscall.SIGTERM == sig || syscall.SIGINT == sig { - // return - // } - // wait for second signal or no more handlers sig = nil for sig == nil && numHandlers != 0 {

1 0

[snowflake/master] log traffic bytes only once every few seconds, along with OnMessage & datachannel.Send counts, to prevent flooded logs
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit c4215b5614210ec81d728d834f42754efb6d5130 Author: Serene Han <keroserene+git(a)gmail.com> Date: Thu Feb 18 14:15:22 2016 -0800 log traffic bytes only once every few seconds, along with OnMessage & datachannel.Send counts, to prevent flooded logs --- client/client_test.go | 4 +++ client/snowflake.go | 17 +++++++++---- client/util.go | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 5 deletions(-) diff --git a/client/client_test.go b/client/client_test.go index ff15d6d..6ee36d9 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -24,6 +24,10 @@ func TestConnect(t *testing.T) { Convey("WebRTC Connection", func() { c := new(webRTCConn) + c.BytesInfo = &BytesInfo{ + inboundChan: make(chan int), outboundChan: make(chan int), + inbound: 0, outbound: 0, inEvents: 0, outEvents: 0, + } So(c.buffer.Bytes(), ShouldEqual, nil) Convey("SendData buffers when datachannel is nil", func() { diff --git a/client/snowflake.go b/client/snowflake.go index 081cb3b..eba68eb 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -67,6 +67,7 @@ type webRTCConn struct { writePipe *io.PipeWriter buffer bytes.Buffer reset chan struct{} + *BytesInfo } var webrtcRemote *webRTCConn @@ -164,12 +165,12 @@ func (c *webRTCConn) EstablishDataChannel() error { dc.OnOpen = func() { log.Println("WebRTC: DataChannel.OnOpen") // if nil != c.snowflake { - // panic("PeerConnection snowflake already exists.") + // panic("PeerConnection snowflake already exists.") // } // Flush the buffer, then enable datachannel. // TODO: Make this more safe dc.Send(c.buffer.Bytes()) - log.Println("Flushed ", c.buffer.Len(), " bytes") + log.Println("Flushed", c.buffer.Len(), "bytes") c.buffer.Reset() c.snowflake = dc } @@ -184,7 +185,7 @@ func (c *webRTCConn) EstablishDataChannel() error { } } dc.OnMessage = func(msg []byte) { - log.Printf("OnMessage <--- %d bytes", len(msg)) + c.BytesInfo.AddInbound(len(msg)) n, err := c.writePipe.Write(msg) if err != nil { // TODO: Maybe shouldn't actually close. @@ -247,6 +248,7 @@ func (c *webRTCConn) ReceiveAnswer() { } func (c *webRTCConn) sendData(data []byte) { + c.BytesInfo.AddOutbound(len(data)) // Buffer the data in case datachannel isn't available yet. if nil == c.snowflake { log.Printf("Buffered %d bytes --> WebRTC", len(data)) @@ -256,10 +258,9 @@ func (c *webRTCConn) sendData(data []byte) { // Otherwise, flush buffer if necessary. for c.buffer.Len() > 0 { c.snowflake.Send(c.buffer.Bytes()) - log.Println("Flushed ", c.buffer.Len(), " bytes") + log.Println("Flushed", c.buffer.Len(), "bytes") c.buffer.Reset() } - log.Printf("Write %d bytes --> WebRTC", len(data)) c.snowflake.Send(data) } @@ -297,6 +298,12 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( connection.answerChannel = make(chan *webrtc.SessionDescription) connection.errorChannel = make(chan error) connection.reset = make(chan struct{}) + connection.BytesInfo = &BytesInfo{ + inboundChan: make(chan int), outboundChan: make(chan int), + inbound: 0, outbound: 0, inEvents: 0, outEvents: 0, + } + go connection.BytesInfo.Log() + // Pipes remain the same even when DataChannel gets switched. connection.recvPipe, connection.writePipe = io.Pipe() diff --git a/client/util.go b/client/util.go new file mode 100644 index 0000000..fa04220 --- /dev/null +++ b/client/util.go @@ -0,0 +1,67 @@ +package main + +import ( + "log" + "time" +) + +type BytesInfo struct { + outboundChan chan int + inboundChan chan int + outbound int + inbound int + outEvents int + inEvents int + isLogging bool +} + +func (b *BytesInfo) Log() { + b.isLogging = true + var amount int + output := func() { + log.Printf("Traffic Bytes (in|out): %d | %d -- (%d OnMessages, %d Sends)", + b.inbound, b.outbound, b.inEvents, b.outEvents) + b.outbound = 0 + b.outEvents = 0 + b.inbound = 0 + b.inEvents = 0 + } + last := time.Now() + for { + select { + case amount = <-b.outboundChan: + b.outbound += amount + b.outEvents++ + last := time.Now() + if time.Since(last) > time.Second*5 { + last = time.Now() + output() + } + case amount = <-b.inboundChan: + b.inbound += amount + b.inEvents++ + if time.Since(last) > time.Second*5 { + last = time.Now() + output() + } + case <-time.After(time.Second * 5): + if b.inEvents > 0 || b.outEvents > 0 { + output() + } + } + } +} + +func (b *BytesInfo) AddOutbound(amount int) { + if !b.isLogging { + return + } + b.outboundChan <- amount +} + +func (b *BytesInfo) AddInbound(amount int) { + if !b.isLogging { + return + } + b.inboundChan <- amount +}

1 0

[snowflake/master] Use reconnect timeout on client (#12)
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit 8a2e7a6041f367a86e3c75df39b010c34649d3b7 Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 19:53:11 2016 -0800 Use reconnect timeout on client (#12) --- client/client_test.go | 7 +++++++ client/rendezvous.go | 4 ++++ client/snowflake.go | 50 +++++++++++++++++++++++++------------------------- 3 files changed, 36 insertions(+), 25 deletions(-) diff --git a/client/client_test.go b/client/client_test.go index 399549a..546382f 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -39,6 +39,13 @@ func TestConnect(t *testing.T) { So(mock.destination.Bytes(), ShouldResemble, []byte("test")) }) + Convey("Receive answer fails on nil answer", func() { + c.reset = make(chan struct{}) + c.ReceiveAnswer() + answerChannel <- nil + <-c.reset + }) + Convey("Connect Loop", func() { // TODO }) diff --git a/client/rendezvous.go b/client/rendezvous.go index ea0b338..fe75b09 100644 --- a/client/rendezvous.go +++ b/client/rendezvous.go @@ -5,6 +5,7 @@ package main import ( "bytes" + "errors" "io/ioutil" "log" "net/http" @@ -66,6 +67,9 @@ func (bc *BrokerChannel) Negotiate(offer *webrtc.SessionDescription) ( } defer resp.Body.Close() log.Printf("BrokerChannel Response:\n%s\n\n", resp.Status) + if http.StatusOK != resp.StatusCode { + return nil, errors.New("no answer from broker.") + } body, err := ioutil.ReadAll(resp.Body) if nil != err { return nil, err diff --git a/client/snowflake.go b/client/snowflake.go index 907c8ae..6d612ce 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -58,7 +58,7 @@ type SnowflakeChannel interface { // Implements net.Conn interface type webRTCConn struct { - config *webrtc.Configuration + config *webrtc.Configuration pc *webrtc.PeerConnection snowflake SnowflakeChannel // Interface holding the WebRTC DataChannel. broker *BrokerChannel @@ -108,7 +108,6 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error { func (c *webRTCConn) PreparePeerConnection() { if nil != c.pc { - log.Printf("PeerConnection already exists.") c.pc.Close() c.pc = nil } @@ -135,12 +134,12 @@ func (c *webRTCConn) PreparePeerConnection() { }() } pc.OnIceCandidate = func(candidate webrtc.IceCandidate) { - log.Printf("OnIceCandidate %s", candidate.Serialize()) + log.Printf("WebRTC: OnIceCandidate %s", candidate.Serialize()) // Allow candidates to accumulate until OnIceComplete. } // TODO: This may soon be deprecated, consider OnIceGatheringStateChange. pc.OnIceComplete = func() { - log.Printf("OnIceComplete") + log.Printf("WebRTC: OnIceComplete") c.offerChannel <- pc.LocalDescription() } // This callback is not expected, as the Client initiates the creation @@ -150,6 +149,7 @@ func (c *webRTCConn) PreparePeerConnection() { panic("Unexpected OnDataChannel!") } c.pc = pc + log.Println("WebRTC: PeerConnection created.") } // Create a WebRTC DataChannel locally. @@ -189,6 +189,7 @@ func (c *webRTCConn) EstablishDataChannel() error { panic("short write") } } + log.Println("WebRTC: DataChannel created.") return nil } @@ -209,15 +210,9 @@ func (c *webRTCConn) SendOffer() error { log.Println("Sending offer via BrokerChannel...\nTarget URL: ", brokerURL, "\nFront URL: ", frontDomain) answer, err := c.broker.Negotiate(c.pc.LocalDescription()) - if nil != err { - log.Printf("BrokerChannel signaling error: %s", err) - return - } - if nil == answer { - log.Printf("BrokerChannel: No answer received.") - // TODO: Should try again here. - c.reset <- struct{}{} - return + if nil != err || nil == answer { + log.Printf("BrokerChannel error: %s", err) + answer = nil } answerChannel <- answer }() @@ -228,17 +223,21 @@ func (c *webRTCConn) SendOffer() error { return nil } -func (c *webRTCConn) ReceiveAnswer() error { - log.Printf("waiting for answer...") - answer, ok := <-answerChannel - if !ok { - // TODO: Don't just fail, try again! - c.pc.Close() - // connection.errorChannel <- errors.New("Bad answer") - return errors.New("Bad answer") - } - log.Printf("Received Answer:\n\n%s\n", answer.Sdp) - return c.pc.SetRemoteDescription(answer) +func (c *webRTCConn) ReceiveAnswer() { + go func() { + answer, ok := <-answerChannel + if !ok || nil == answer { + log.Printf("Failed to retrieve answer. Retrying in %d seconds", ReconnectTimeout) + <-time.After(time.Second * ReconnectTimeout) + c.reset <- struct{}{} + return + } + log.Printf("Received Answer:\n\n%s\n", answer.Sdp) + err := c.pc.SetRemoteDescription(answer) + if nil != err { + c.errorChannel <- err + } + }() } func (c *webRTCConn) sendData(data []byte) { @@ -259,10 +258,11 @@ func (c *webRTCConn) ConnectLoop() { // TODO: When go-webrtc is more stable, it's possible that a new // PeerConnection won't need to be recreated each time. // called once. - c.PreparePeerConnection() + c.PreparePeerConnection() c.EstablishDataChannel() c.SendOffer() c.ReceiveAnswer() + <-c.reset log.Println(" --- snowflake connection reset ---") }

1 0

[snowflake/master] Include answer channel as part of the webRTCConn struct (#12)
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit a1b7e01c5423c69548afdf44c7455736605b5574 Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 20:41:33 2016 -0800 Include answer channel as part of the webRTCConn struct (#12) --- client/client_test.go | 4 ++- client/snowflake.go | 78 ++++++++++++++++++++++++++++++++------------------- 2 files changed, 52 insertions(+), 30 deletions(-) diff --git a/client/client_test.go b/client/client_test.go index 546382f..ff15d6d 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "github.com/keroserene/go-webrtc" . "github.com/smartystreets/goconvey/convey" "testing" ) @@ -41,8 +42,9 @@ func TestConnect(t *testing.T) { Convey("Receive answer fails on nil answer", func() { c.reset = make(chan struct{}) + c.answerChannel = make(chan *webrtc.SessionDescription) c.ReceiveAnswer() - answerChannel <- nil + c.answerChannel <- nil <-c.reset }) diff --git a/client/snowflake.go b/client/snowflake.go index 6d612ce..081cb3b 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -28,7 +28,6 @@ var frontDomain string // When a connection handler starts, +1 is written to this channel; when it // ends, -1 is written. var handlerChan = make(chan int) -var answerChannel = make(chan *webrtc.SessionDescription) const ( ReconnectTimeout = 5 @@ -37,7 +36,6 @@ const ( func copyLoop(a, b net.Conn) { var wg sync.WaitGroup wg.Add(2) - go func() { io.Copy(b, a) wg.Done() @@ -46,8 +44,8 @@ func copyLoop(a, b net.Conn) { io.Copy(a, b) wg.Done() }() - wg.Wait() + log.Println("copy loop ended") } // Interface that matches both webrc.DataChannel and for testing. @@ -58,16 +56,17 @@ type SnowflakeChannel interface { // Implements net.Conn interface type webRTCConn struct { - config *webrtc.Configuration - pc *webrtc.PeerConnection - snowflake SnowflakeChannel // Interface holding the WebRTC DataChannel. - broker *BrokerChannel - offerChannel chan *webrtc.SessionDescription - errorChannel chan error - recvPipe *io.PipeReader - writePipe *io.PipeWriter - buffer bytes.Buffer - reset chan struct{} + config *webrtc.Configuration + pc *webrtc.PeerConnection + snowflake SnowflakeChannel // Interface holding the WebRTC DataChannel. + broker *BrokerChannel + offerChannel chan *webrtc.SessionDescription + answerChannel chan *webrtc.SessionDescription + errorChannel chan error + recvPipe *io.PipeReader + writePipe *io.PipeWriter + buffer bytes.Buffer + reset chan struct{} } var webrtcRemote *webRTCConn @@ -164,6 +163,9 @@ func (c *webRTCConn) EstablishDataChannel() error { } dc.OnOpen = func() { log.Println("WebRTC: DataChannel.OnOpen") + // if nil != c.snowflake { + // panic("PeerConnection snowflake already exists.") + // } // Flush the buffer, then enable datachannel. // TODO: Make this more safe dc.Send(c.buffer.Bytes()) @@ -175,14 +177,18 @@ func (c *webRTCConn) EstablishDataChannel() error { // Disable the DataChannel as a write destination. // Future writes will go to the buffer until a new DataChannel is available. log.Println("WebRTC: DataChannel.OnClose") - c.snowflake = nil - c.reset <- struct{}{} // Attempt to negotiate a new datachannel.. + if nil != c.snowflake { + c.snowflake = nil + // Only reset if this OnClose triggered + c.Reset() + } } dc.OnMessage = func(msg []byte) { log.Printf("OnMessage <--- %d bytes", len(msg)) n, err := c.writePipe.Write(msg) if err != nil { // TODO: Maybe shouldn't actually close. + log.Println("Error writing to SOCKS pipe") c.writePipe.CloseWithError(err) } if n != len(msg) { @@ -205,7 +211,7 @@ func (c *webRTCConn) SendOffer() error { log.Printf("----------------") return nil } - // Use Broker... + // Otherwise, use Broker. go func() { log.Println("Sending offer via BrokerChannel...\nTarget URL: ", brokerURL, "\nFront URL: ", frontDomain) @@ -214,7 +220,7 @@ func (c *webRTCConn) SendOffer() error { log.Printf("BrokerChannel error: %s", err) answer = nil } - answerChannel <- answer + c.answerChannel <- answer }() case err := <-c.errorChannel: c.pc.Close() @@ -225,11 +231,11 @@ func (c *webRTCConn) SendOffer() error { func (c *webRTCConn) ReceiveAnswer() { go func() { - answer, ok := <-answerChannel + answer, ok := <-c.answerChannel if !ok || nil == answer { log.Printf("Failed to retrieve answer. Retrying in %d seconds", ReconnectTimeout) <-time.After(time.Second * ReconnectTimeout) - c.reset <- struct{}{} + c.Reset() return } log.Printf("Received Answer:\n\n%s\n", answer.Sdp) @@ -247,6 +253,12 @@ func (c *webRTCConn) sendData(data []byte) { c.buffer.Write(data) return } + // Otherwise, flush buffer if necessary. + for c.buffer.Len() > 0 { + c.snowflake.Send(c.buffer.Bytes()) + log.Println("Flushed ", c.buffer.Len(), " bytes") + c.buffer.Reset() + } log.Printf("Write %d bytes --> WebRTC", len(data)) c.snowflake.Send(data) } @@ -256,18 +268,25 @@ func (c *webRTCConn) ConnectLoop() { for { log.Println("Establishing WebRTC connection...") // TODO: When go-webrtc is more stable, it's possible that a new - // PeerConnection won't need to be recreated each time. - // called once. + // PeerConnection won't need to be re-prepared each time. c.PreparePeerConnection() - c.EstablishDataChannel() - c.SendOffer() - c.ReceiveAnswer() - - <-c.reset - log.Println(" --- snowflake connection reset ---") + err := c.EstablishDataChannel() + if err == nil { + c.SendOffer() + c.ReceiveAnswer() + <-c.reset + log.Println(" --- snowflake connection reset ---") + } } } +func (c *webRTCConn) Reset() { + go func() { + c.reset <- struct{}{} // Attempt to negotiate a new datachannel.. + log.Println("WebRTC resetting...") + }() +} + // Initialize a WebRTC Connection. func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( *webRTCConn, error) { @@ -275,6 +294,7 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( connection.config = config connection.broker = broker connection.offerChannel = make(chan *webrtc.SessionDescription) + connection.answerChannel = make(chan *webrtc.SessionDescription) connection.errorChannel = make(chan error) connection.reset = make(chan struct{}) // Pipes remain the same even when DataChannel gets switched. @@ -363,10 +383,10 @@ func readSignalingMessages(f *os.File) { log.Printf("ignoring invalid signal message %+q", msg) continue } - answerChannel <- sdp + webrtcRemote.answerChannel <- sdp } log.Printf("close answerChannel") - close(answerChannel) + close(webrtcRemote.answerChannel) if err := s.Err(); err != nil { log.Printf("signal FIFO: %s", err) }

1 0

[snowflake/master] Snowflake client now using a reconnect loop (#12)
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit f205a0be59aec40a04580cd46dad1bf6e9eba6c3 Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 19:19:11 2016 -0800 Snowflake client now using a reconnect loop (#12) --- client/client_test.go | 4 ++ client/snowflake.go | 151 +++++++++++++++++++++++++++++--------------------- 2 files changed, 93 insertions(+), 62 deletions(-) diff --git a/client/client_test.go b/client/client_test.go index 7b8dad2..399549a 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -38,6 +38,10 @@ func TestConnect(t *testing.T) { So(c.buffer.Bytes(), ShouldEqual, nil) So(mock.destination.Bytes(), ShouldResemble, []byte("test")) }) + + Convey("Connect Loop", func() { + // TODO + }) }) }) diff --git a/client/snowflake.go b/client/snowflake.go index 771e90b..907c8ae 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -28,9 +28,12 @@ var frontDomain string // When a connection handler starts, +1 is written to this channel; when it // ends, -1 is written. var handlerChan = make(chan int) - var answerChannel = make(chan *webrtc.SessionDescription) +const ( + ReconnectTimeout = 5 +) + func copyLoop(a, b net.Conn) { var wg sync.WaitGroup wg.Add(2) @@ -55,14 +58,16 @@ type SnowflakeChannel interface { // Implements net.Conn interface type webRTCConn struct { + config *webrtc.Configuration pc *webrtc.PeerConnection - snowflake SnowflakeChannel // Interface holding the WebRTC DataChannel. + snowflake SnowflakeChannel // Interface holding the WebRTC DataChannel. broker *BrokerChannel offerChannel chan *webrtc.SessionDescription errorChannel chan error recvPipe *io.PipeReader writePipe *io.PipeWriter buffer bytes.Buffer + reset chan struct{} } var webrtcRemote *webRTCConn @@ -72,7 +77,6 @@ func (c *webRTCConn) Read(b []byte) (int, error) { } func (c *webRTCConn) Write(b []byte) (int, error) { - // log.Printf("webrtc Write %d %+q", len(b), string(b)) c.sendData(b) return len(b), nil } @@ -102,10 +106,56 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error { return fmt.Errorf("SetWriteDeadline not implemented") } +func (c *webRTCConn) PreparePeerConnection() { + if nil != c.pc { + log.Printf("PeerConnection already exists.") + c.pc.Close() + c.pc = nil + } + pc, err := webrtc.NewPeerConnection(c.config) + if err != nil { + log.Printf("NewPeerConnection: %s", err) + c.errorChannel <- err + } + // Prepare PeerConnection callbacks. + pc.OnNegotiationNeeded = func() { + log.Println("WebRTC: OnNegotiationNeeded") + go func() { + offer, err := pc.CreateOffer() + // TODO: Potentially timeout and retry if ICE isn't working. + if err != nil { + c.errorChannel <- err + return + } + err = pc.SetLocalDescription(offer) + if err != nil { + c.errorChannel <- err + return + } + }() + } + pc.OnIceCandidate = func(candidate webrtc.IceCandidate) { + log.Printf("OnIceCandidate %s", candidate.Serialize()) + // Allow candidates to accumulate until OnIceComplete. + } + // TODO: This may soon be deprecated, consider OnIceGatheringStateChange. + pc.OnIceComplete = func() { + log.Printf("OnIceComplete") + c.offerChannel <- pc.LocalDescription() + } + // This callback is not expected, as the Client initiates the creation + // of the data channel, not the remote peer. + pc.OnDataChannel = func(channel *webrtc.DataChannel) { + log.Println("OnDataChannel") + panic("Unexpected OnDataChannel!") + } + c.pc = pc +} + // Create a WebRTC DataChannel locally. func (c *webRTCConn) EstablishDataChannel() error { dc, err := c.pc.CreateDataChannel("snowflake", webrtc.Init{}) - // Triggers "OnNegotiationNeeded" on the PeerConnection, which will prepare + // Triggers "OnNegotiationNeeded" on the PeerConnection, which will prepare // an SDP offer while other goroutines operating on this struct handle the // signaling. Eventually fires "OnOpen". if err != nil { @@ -126,7 +176,7 @@ func (c *webRTCConn) EstablishDataChannel() error { // Future writes will go to the buffer until a new DataChannel is available. log.Println("WebRTC: DataChannel.OnClose") c.snowflake = nil - // TODO: (Issue #12) Should attempt to renegotiate at this point. + c.reset <- struct{}{} // Attempt to negotiate a new datachannel.. } dc.OnMessage = func(msg []byte) { log.Printf("OnMessage <--- %d bytes", len(msg)) @@ -144,7 +194,7 @@ func (c *webRTCConn) EstablishDataChannel() error { // Block until an offer is available, then send it to either // the Broker or signal pipe. -func (c *webRTCConn) sendOffer() error { +func (c *webRTCConn) SendOffer() error { select { case offer := <-c.offerChannel: if "" == brokerURL { @@ -166,6 +216,7 @@ func (c *webRTCConn) sendOffer() error { if nil == answer { log.Printf("BrokerChannel: No answer received.") // TODO: Should try again here. + c.reset <- struct{}{} return } answerChannel <- answer @@ -177,6 +228,19 @@ func (c *webRTCConn) sendOffer() error { return nil } +func (c *webRTCConn) ReceiveAnswer() error { + log.Printf("waiting for answer...") + answer, ok := <-answerChannel + if !ok { + // TODO: Don't just fail, try again! + c.pc.Close() + // connection.errorChannel <- errors.New("Bad answer") + return errors.New("Bad answer") + } + log.Printf("Received Answer:\n\n%s\n", answer.Sdp) + return c.pc.SetRemoteDescription(answer) +} + func (c *webRTCConn) sendData(data []byte) { // Buffer the data in case datachannel isn't available yet. if nil == c.snowflake { @@ -188,72 +252,35 @@ func (c *webRTCConn) sendData(data []byte) { c.snowflake.Send(data) } +// WebRTC re-establishment loop. Expected in own goroutine. +func (c *webRTCConn) ConnectLoop() { + for { + log.Println("Establishing WebRTC connection...") + // TODO: When go-webrtc is more stable, it's possible that a new + // PeerConnection won't need to be recreated each time. + // called once. + c.PreparePeerConnection() + c.EstablishDataChannel() + c.SendOffer() + c.ReceiveAnswer() + <-c.reset + log.Println(" --- snowflake connection reset ---") + } +} + // Initialize a WebRTC Connection. func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( *webRTCConn, error) { - pc, err := webrtc.NewPeerConnection(config) - if err != nil { - log.Printf("NewPeerConnection: %s", err) - return nil, err - } connection := new(webRTCConn) + connection.config = config connection.broker = broker - connection.pc = pc connection.offerChannel = make(chan *webrtc.SessionDescription) connection.errorChannel = make(chan error) + connection.reset = make(chan struct{}) // Pipes remain the same even when DataChannel gets switched. connection.recvPipe, connection.writePipe = io.Pipe() - pc.OnNegotiationNeeded = func() { - log.Println("OnNegotiationNeeded") - go func() { - offer, err := pc.CreateOffer() - // TODO: Potentially timeout and retry if ICE isn't working. - if err != nil { - connection.errorChannel <- err - return - } - err = pc.SetLocalDescription(offer) - if err != nil { - connection.errorChannel <- err - return - } - }() - } - pc.OnIceCandidate = func(candidate webrtc.IceCandidate) { - log.Printf("OnIceCandidate %s", candidate.Serialize()) - // Allow candidates to accumulate until OnIceComplete. - } - // TODO: This may soon be deprecated, consider OnIceGatheringStateChange. - pc.OnIceComplete = func() { - log.Printf("OnIceComplete") - connection.offerChannel <- pc.LocalDescription() - } - // This callback is not expected, as the Client initiates the creation - // of the data channel, not the remote peer. - pc.OnDataChannel = func(channel *webrtc.DataChannel) { - log.Println("OnDataChannel") - panic("Unexpected OnDataChannel!") - } - - connection.EstablishDataChannel() - - // TODO: Make this part of a re-establishment loop. - connection.sendOffer() - log.Printf("waiting for answer...") - answer, ok := <-answerChannel - if !ok { - // TODO: Don't just fail, try again! - pc.Close() - return nil, fmt.Errorf("no answer received") - } - log.Printf("Received Answer:\n\n%s\n", answer.Sdp) - err = pc.SetRemoteDescription(answer) - if err != nil { - pc.Close() - return nil, err - } - + go connection.ConnectLoop() return connection, nil }

1 0

[snowflake/master] Disconnections from websocket side also resets browser proxy
by serene＠torproject.org 18 Feb '16

18 Feb '16

commit e35687b587f219e8c1b1323e09765b3ec85daf69 Author: Serene Han <keroserene+git(a)gmail.com> Date: Wed Feb 17 19:55:43 2016 -0800 Disconnections from websocket side also resets browser proxy --- proxy/proxypair.coffee | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/proxy/proxypair.coffee b/proxy/proxypair.coffee index e7adfb9..b8db513 100644 --- a/proxy/proxypair.coffee +++ b/proxy/proxypair.coffee @@ -63,7 +63,7 @@ class ProxyPair @connectRelay() channel.onclose = -> log 'WebRTC DataChannel closed.' - snowflake.ui.setStatus 'disconnected.' + snowflake.ui.setStatus 'disconnected by webrtc.' snowflake.ui.setActive false snowflake.state = MODE.INIT # Change this for multiplexing. @@ -105,6 +105,9 @@ class ProxyPair onClose: (event) => ws = event.target log ws.label + ' closed.' + snowflake.ui.setStatus 'disconnected.' + snowflake.ui.setActive false + snowflake.state = MODE.INIT @flush() @maybeCleanup()

1 0