HTTP handlers implemented

Author: AndrewBurian

client.go

@@ -35,6 +35,12 @@ func NewClient(w http.ResponseWriter) *Client {
 	}
 	c.close = closer
 
+	// Send the initial headers
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	flush.Flush()
+
 	// start the sending thread
 	c.waiter.Add(1)
 	go c.run()

stream.go

@@ -18,11 +18,6 @@ type registeredClient struct {
 	topics map[string]bool
 }
 
-func New() *Stream {
-	s := &Stream{}
-	return s
-}
-
 // Register adds a client to the stream to receive all broadcast
 // messages. Has no effect if the client is already registered.
 func (s *Stream) Register(c *Client) {
@@ -95,12 +90,57 @@ func (s *Stream) CloseTopic(topic string) {
 
 }
 
+// ServeHTTP takes a client connection, registers it for broadcasts,
+// then blocks so long as the connection is alive.
 func (s *Stream) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
+	// ensure the client accepts an event-stream
+	if !checkRequest(r) {
+		http.Error(w, "This is an EventStream endpoint", http.StatusNotAcceptable)
+		return
+	}
+
+	// create the client
+	c := NewClient(w)
+	if c == nil {
+		http.Error(w, "EventStream not supported for this connection", http.StatusInternalServerError)
+		return
+	}
+
+	// wait for the client to exit or be shutdown
+	s.Register(c)
+	c.Wait()
 }
 
-func (s *Stream) TopicHandler(topic string) http.HandlerFunc {
+// TopicHandler returns an HTTP handler that will register a client for broadcasts
+// and for any topics, and then block so long as they are connected
+func (s *Stream) TopicHandler(topics []string) http.HandlerFunc {
 
+	return func(w http.ResponseWriter, r *http.Request) {
+		// ensure the client accepts an event-stream
+		if !checkRequest(r) {
+			http.Error(w, "This is an EventStream endpoint", http.StatusNotAcceptable)
+			return
+		}
+
+		// create the client
+		c := NewClient(w)
+		if c == nil {
+			http.Error(w, "EventStream not supported for this connection", http.StatusInternalServerError)
+			return
+		}
+
+		// broadcasts
+		s.Register(c)
+
+		// topics
+		for _, topic := range topics {
+			s.Subscribe(topic, c)
+		}
+
+		// wait for the client to exit or be shutdown
+		c.Wait()
+	}
 }
 
 // Register a function to be called when a client connects to this stream's
@@ -109,8 +149,9 @@ func (s *Stream) ClientConnectHook(fn func(*http.Request, *Client)) {
 	s.clientConnectHook = fn
 }
 
-func (s *Stream) sendAll(ev *Event) {
-
+// Checks that a client expects an event-stream
+func checkRequest(r *http.Request) bool {
+	return r.Header.Get("Accept") == "text/event-stream"
 }
 
 func (s *Stream) getClient(c *Client) *registeredClient {
@@ -130,12 +171,15 @@ func (s *Stream) getClient(c *Client) *registeredClient {
 }
 
 func (s *Stream) addClient(c *Client) *registeredClient {
-	s.listLock.Lock()
 
-	s.clients.PushBack(&registeredClient{
+	cli := &registeredClient{
 		c:      c,
 		topics: make(map[string]bool),
-	})
+	}
 
+	s.listLock.Lock()
+	s.clients.PushBack(cli)
 	s.listLock.Unlock()
+
+	return cli
 }