commit 2ae6559001ab779a0c8e9b4b6a6804353b19222a Author: Serene Han <keroserene+git@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