commit 254223be0f7372669a5d42c6ed1bcf1cc559546b Author: Serene Han <keroserene+git@gmail.com> Date: Mon Feb 15 12:23:35 2016 -0800 End-to-end Goconvey test for Broker passing SDP messages between concurrent client and proxy request handlers --- broker/broker.go | 16 +++++----- broker/snowflake-broker_test.go | 66 ++++++++++++++++++++++++++++++++++++----- 2 files changed, 66 insertions(+), 16 deletions(-) diff --git a/broker/broker.go b/broker/broker.go index 7c14f63..69b8369 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 - createChan chan *ProxyRequest + snowflakeMap map[string]*Snowflake + createChannel chan *ProxyRequest } func NewBrokerContext() *BrokerContext { snowflakes := new(SnowflakeHeap) heap.Init(snowflakes) return &BrokerContext{ - snowflakes: snowflakes, - snowflakeMap: make(map[string]*Snowflake), - createChan: make(chan *ProxyRequest), + snowflakes: snowflakes, + snowflakeMap: make(map[string]*Snowflake), + createChannel: make(chan *ProxyRequest), } } @@ -72,7 +72,7 @@ func (sc *BrokerContext) AddSnowflake(id string) *Snowflake { // func (ctx *BrokerContext) Broker(proxies <-chan *ProxyRequest) { func (ctx *BrokerContext) Broker() { // for p := range proxies { - for p := range ctx.createChan { + 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. @@ -131,7 +131,7 @@ func clientHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { 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' + // TODO: Needs improvement - maybe shouldn't immediately fail? if ctx.snowflakes.Len() <= 0 { log.Println("Client: No snowflake proxies available.") w.WriteHeader(http.StatusServiceUnavailable) @@ -178,7 +178,7 @@ func proxyHandler(ctx *BrokerContext, w http.ResponseWriter, r *http.Request) { p := new(ProxyRequest) p.id = id p.offerChan = make(chan []byte) - ctx.createChan <- p + ctx.createChannel <- p // Wait for a client to avail an offer to the snowflake, or timeout // and ask the snowflake to poll later. diff --git a/broker/snowflake-broker_test.go b/broker/snowflake-broker_test.go index 06d2316..ee984b0 100644 --- a/broker/snowflake-broker_test.go +++ b/broker/snowflake-broker_test.go @@ -7,7 +7,6 @@ import ( "net/http" "net/http/httptest" "testing" - "fmt" ) func TestBroker(t *testing.T) { @@ -84,7 +83,7 @@ func TestBroker(t *testing.T) { done <- true }(ctx) // Pass a fake client offer to this proxy - p := <-ctx.createChan + p := <-ctx.createChannel So(p.id, ShouldEqual, "test") p.offerChan <- []byte("fake offer") <-done @@ -97,7 +96,7 @@ func TestBroker(t *testing.T) { proxyHandler(ctx, w, r) done <- true }(ctx) - p := <-ctx.createChan + p := <-ctx.createChannel So(p.id, ShouldEqual, "test") // nil means timeout p.offerChan <- nil @@ -107,10 +106,10 @@ func TestBroker(t *testing.T) { }) }) - Convey("Responds to proxy answers...", func() { + Convey("Responds to proxy answers...", func() { + s := ctx.AddSnowflake("test") w := httptest.NewRecorder() data := bytes.NewReader([]byte("fake answer")) - s := ctx.AddSnowflake("test") Convey("by passing to the client if valid.", func() { r, err := http.NewRequest("POST", "snowflake.broker/answer", data) @@ -119,7 +118,7 @@ func TestBroker(t *testing.T) { go func(ctx *BrokerContext) { answerHandler(ctx, w, r) }(ctx) - answer := <- s.answerChannel + answer := <-s.answerChannel So(w.Code, ShouldEqual, http.StatusOK) So(answer, ShouldResemble, []byte("fake answer")) }) @@ -130,7 +129,6 @@ func TestBroker(t *testing.T) { r.Header.Set("X-Session-ID", "invalid") answerHandler(ctx, w, r) So(w.Code, ShouldEqual, http.StatusGone) - fmt.Println("omg") }) Convey("with error if the proxy gives invalid answer", func() { @@ -141,9 +139,61 @@ func TestBroker(t *testing.T) { answerHandler(ctx, w, r) So(w.Code, ShouldEqual, http.StatusBadRequest) }) - }) + }) + + Convey("End-To-End", t, func() { + done := make(chan bool) + polled := make(chan bool) + ctx := NewBrokerContext() + // Proxy polls with its ID first... + dataP := bytes.NewReader([]byte("test")) + wP := httptest.NewRecorder() + rP, err := http.NewRequest("POST", "snowflake.broker/proxy", dataP) + So(err, ShouldBeNil) + rP.Header.Set("X-Session-ID", "test") + go func() { + proxyHandler(ctx, wP, rP) + polled <- true + }() + + // Manually do the Broker goroutine action here for full control. + p := <-ctx.createChannel + So(p.id, ShouldEqual, "test") + s := ctx.AddSnowflake(p.id) + go func() { + offer := <-s.offerChannel + p.offerChan <- offer + }() + So(ctx.snowflakeMap["test"], ShouldNotBeNil) + + // Client request blocks until proxy answer arrives. + dataC := bytes.NewReader([]byte("fake offer")) + wC := httptest.NewRecorder() + rC, err := http.NewRequest("POST", "snowflake.broker/client", dataC) + So(err, ShouldBeNil) + go func() { + clientHandler(ctx, wC, rC) + done <- true + }() + + <-polled + So(wP.Code, ShouldEqual, http.StatusOK) + So(wP.Body.String(), ShouldResemble, "fake offer") + So(ctx.snowflakeMap["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) + So(wA.Code, ShouldEqual, http.StatusOK) + + <-done + So(wC.Code, ShouldEqual, http.StatusOK) + So(wC.Body.String(), ShouldEqual, "fake answer") }) }