broker tracking a heap of snowflakes

broker/snowflake-broker.go

@@ -1,13 +1,13 @@
 package snowflake_broker
 
 import (
-	// "io"
+	"container/heap"
+	"fmt"
 	"io/ioutil"
 	"log"
 	"net"
 	"net/http"
-	"path"
-
+	"time"
 	// "appengine"
 	// "appengine/urlfetch"
 )
@@ -15,7 +15,64 @@ import (
 // This is an intermediate step - a basic hardcoded appengine rendezvous
 // to a single browser snowflake.
 
-var snowflakeProxy = ""
+// This is minimum viable client-proxy registration.
+// TODO: better, more secure registration corresponding to what's in
+// the python flashproxy facilitator.
+
+// Slice of available snowflake proxies.
+// var snowflakes []chan []byte
+
+type Snowflake struct {
+	id         string
+	sigChannel chan []byte
+	clients    int
+	index      int
+}
+
+// Implements heap.Interface, and holds Snowflakes.
+type SnowflakeHeap []*Snowflake
+
+func (sh SnowflakeHeap) Len() int { return len(sh) }
+
+func (sh SnowflakeHeap) Less(i, j int) bool {
+	// Snowflakes serving less clients should sort earlier.
+	return sh[i].clients < sh[j].clients
+}
+
+func (sh SnowflakeHeap) Swap(i, j int) {
+	sh[i], sh[j] = sh[j], sh[i]
+	sh[i].index = i
+	sh[j].index = j
+}
+
+func (sh *SnowflakeHeap) Push(s interface{}) {
+	n := len(*sh)
+	snowflake := s.(*Snowflake)
+	snowflake.index = n
+	*sh = append(*sh, snowflake)
+}
+
+// Only valid when Len() > 0.
+func (sh *SnowflakeHeap) Pop() interface{} {
+	flakes := *sh
+	n := len(flakes)
+	snowflake := flakes[n-1]
+	snowflake.index = -1
+	*sh = flakes[0 : n-1]
+	return snowflake
+}
+
+var snowflakes *SnowflakeHeap
+
+// Create and add a Snowflake to the heap.
+func AddSnowflake(id string) *Snowflake {
+	snowflake := new(Snowflake)
+	snowflake.id = id
+	snowflake.clients = 0
+	snowflake.sigChannel = make(chan []byte)
+	heap.Push(snowflakes, snowflake)
+	return snowflake
+}
 
 func robotsTxtHandler(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
@@ -36,29 +93,76 @@ 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 regHandler(w http.ResponseWriter, r *http.Request) {
-	// TODO: Maybe don't pass anything on path, since it will always be bidirectional
-	dir, _ := path.Split(path.Clean(r.URL.Path))
-	if dir != "/reg/" {
-		http.NotFound(w, r)
-		return
-	}
-	body, err := ioutil.ReadAll(r.Body)
+func clientHandler(w http.ResponseWriter, r *http.Request) {
+	offer, err := ioutil.ReadAll(r.Body)
 	if nil != err {
-		return
 		log.Println("Invalid data.")
+		return
 	}
 	w.Header().Set("Access-Control-Allow-Origin", "*")
+	// Pop the most available snowflake proxy, and pass the offer to it.
+	// TODO: Make this much better.
+	snowflake := heap.Pop(snowflakes).(*Snowflake)
+	if nil == snowflake {
+		w.Write([]byte("no snowflake proxies available"))
+		// w.WriteHeader(http.StatusServiceUnavailable)
+		return
+	}
+	// snowflakes = snowflakes[1:]
+	snowflake.sigChannel <- offer
+	w.Write([]byte("sent offer to proxy!"))
 	// TODO: Get browser snowflake to talkto this appengine instance
 	// so it can reply with an answer, and not just the offer again :)
 	// TODO: Real broker which matches clients and snowflake proxies.
+	w.Write(offer)
+}
+
+/*
+A snowflake browser proxy requests a client from the Broker.
+*/
+func proxyHandler(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if nil != err {
+		log.Println("Invalid data.")
+		return
+	}
+	w.Header().Set("Access-Control-Allow-Origin", "*")
+	snowflakeSession := body
+	log.Println("Received snowflake: ", snowflakeSession)
+	snowflake := AddSnowflake(string(snowflakeSession))
+	select {
+	case offer := <-snowflake.sigChannel:
+		log.Println("Passing client offer to snowflake.")
+		w.Write(offer)
+	case <-time.After(time.Second * 10):
+		s := fmt.Sprintf("%d snowflakes left.", snowflakes.Len())
+		w.Write([]byte("timed out. " + s))
+		heap.Remove(snowflakes, snowflake.index)
+		// w.WriteHeader(http.StatusRequestTimeout)
+	}
+}
+
+func reflectHandler(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if nil != err {
+		log.Println("Invalid data.")
+		return
+	}
+	w.Header().Set("Access-Control-Allow-Origin", "*")
 	w.Write(body)
 }
 
 func init() {
+	// snowflakes = make([]chan []byte, 0)
+	snowflakes = new(SnowflakeHeap)
+	heap.Init(snowflakes)
+
 	http.HandleFunc("/robots.txt", robotsTxtHandler)
 	http.HandleFunc("/ip", ipHandler)
-	http.HandleFunc("/reg/", regHandler)
+
+	http.HandleFunc("/client", clientHandler)
+	http.HandleFunc("/proxy", proxyHandler)
+	http.HandleFunc("/reflect", reflectHandler)
 	// if SNOWFLAKE_BROKER == "" {
 	// panic("SNOWFLAKE_BROKER empty; did you forget to edit config.go?")
 	// }

broker/snowflake-broker_test.go

@@ -0,0 +1,59 @@
+package snowflake_broker
+
+import (
+	"container/heap"
+	"testing"
+)
+
+func TestSnowflakeHeap(t *testing.T) {
+	h := new(SnowflakeHeap)
+	heap.Init(h)
+	if 0 != h.Len() {
+		t.Error("Unexpected length.")
+	}
+	s1 := new(Snowflake)
+	s2 := new(Snowflake)
+	s3 := new(Snowflake)
+	s4 := new(Snowflake)
+
+	s1.clients = 4
+	s2.clients = 5
+	s3.clients = 3
+	s4.clients = 1
+
+	heap.Push(h, s1)
+	heap.Push(h, s2)
+	heap.Push(h, s3)
+	heap.Push(h, s4)
+
+	if 4 != h.Len() {
+		t.Error("Unexpected length.")
+	}
+
+	heap.Remove(h, 0)
+	if 3 != h.Len() {
+		t.Error("Unexpected length.")
+	}
+
+	r := heap.Pop(h).(*Snowflake)
+	if r.clients != 3 {
+		t.Error("Unexpected clients: ", r.clients)
+	}
+	if r.index != -1 {
+		t.Error("Unexpected index: ", r.index)
+	}
+
+	r = heap.Pop(h).(*Snowflake)
+	if r.clients != 4 {
+		t.Error("Unexpected clients: ", r.clients)
+	}
+
+	r = heap.Pop(h).(*Snowflake)
+	if r.clients != 5 {
+		t.Error("Unexpected clients: ", r.clients)
+	}
+
+	if 0 != h.Len() {
+		t.Error("Unexpected length.")
+	}
+}