distribute the next ticket via SSE

2 files changed, 142 insertions, 8 deletions

diff --git a/index.html b/index.html
index 51b8a13..6973be9 100644
--- a/index.html
+++ b/index.html
@@ -91,8 +91,14 @@ button {
       }
 
       function finishHelp() {
-        new HttpClient().get("done", () => { location.reload(); } );
+        new HttpClient().get("done", () => {});
       }
+
+      const evtSource = new EventSource("events");
+      evtSource.onmessage = (event) => {
+        let doneNode = document.getElementById("done");
+        doneNode.innerHTML = event.data;
+      };
     </script>
   </head>
   <body>
diff --git a/main.go b/main.go
index 5cb485c..8754dd6 100644
--- a/main.go
+++ b/main.go
@@ -6,6 +6,7 @@ import (
 	"html/template"
 	"log"
 	"net/http"
+	"strconv"
 	"sync/atomic"
 )
 
@@ -17,6 +18,124 @@ var pageTemplate *template.Template
 var open atomic.Uint64
 var done atomic.Uint64
 
+var broker *Broker
+
+// A single Broker will be created in this program. It is responsible
+// for keeping a list of which clients (browsers) are currently attached
+// and broadcasting events (messages) to those clients.
+type Broker struct {
+
+	// Create a map of clients, the keys of the map are the channels
+	// over which we can push messages to attached clients.  (The values
+	// are just booleans and are meaningless.)
+	//
+	clients map[chan string]bool
+
+	// Channel into which new clients can be pushed
+	//
+	newClients chan chan string
+
+	// Channel into which disconnected clients should be pushed
+	//
+	defunctClients chan chan string
+
+	// Channel into which messages are pushed to be broadcast out
+	// to attahed clients.
+	//
+	messages chan string
+}
+
+// This Broker method starts a new goroutine.  It handles
+// the addition & removal of clients, as well as the broadcasting
+// of messages out to clients that are currently attached.
+func (b *Broker) Start() {
+	go func() {
+		for {
+			// Block until we receive from one of the
+			// three following channels.
+			select {
+			case s := <-b.newClients:
+				// There is a new client attached and we
+				// want to start sending them messages.
+				b.clients[s] = true
+				log.Println("Added new client")
+
+			case s := <-b.defunctClients:
+				// A client has dettached and we want to
+				// stop sending them messages.
+				delete(b.clients, s)
+				close(s)
+				log.Println("Removed client")
+
+			case msg := <-b.messages:
+				// There is a new message to send.  For each
+				// attached client, push the new message
+				// into the client's message channel.
+				for s := range b.clients {
+					s <- msg
+				}
+				log.Printf("Broadcast message to %d clients", len(b.clients))
+			}
+		}
+	}()
+}
+
+// This Broker method handles and HTTP request at the "/events" Endpoint.
+func (b *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// Make sure that the writer supports flushing.
+	f, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "Streaming unsupported!", http.StatusInternalServerError)
+		return
+	}
+
+	// Create a new channel, over which the broker can
+	// send this client messages.
+	messageChan := make(chan string)
+
+	// Add this client to the map of those that should
+	// receive updates
+	b.newClients <- messageChan
+
+	// Listen to the closing of the http connection via the Request.Context
+	notify := r.Context().Done()
+	go func() {
+		<-notify
+		// Remove this client from the map of attached clients
+		// when `EventHandler` exits.
+		b.defunctClients <- messageChan
+		log.Println("HTTP connection just closed.")
+	}()
+
+	// Set the headers related to event streaming.
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("Transfer-Encoding", "chunked")
+
+	for {
+		msg, open := <-messageChan
+
+		if !open {
+			// If our messageChan was closed, this means that the client has
+			// disconnected.
+			break
+		}
+
+		if _, err := fmt.Fprintf(w, "data: %s\n\n", msg); err != nil {
+			log.Printf("Writing SSE failed: %v\n", err)
+			break
+		}
+
+		// Flush the response.  This is only possible if
+		// the repsonse supports streaming.
+		f.Flush()
+	}
+
+	// Done.
+	log.Println("Finished HTTP request at ", r.URL.Path)
+}
+
 func pageHandler(w http.ResponseWriter, r *http.Request) {
 	if r.URL.Path != "/" {
 		w.WriteHeader(http.StatusNotFound)
@@ -46,22 +165,31 @@ func newHandler(w http.ResponseWriter, r *http.Request) {
 func doneHandler(w http.ResponseWriter, r *http.Request) {
 	ticket := done.Add(1)
 	log.Printf("Finished ticket %d\n", ticket)
+	broker.messages <- strconv.FormatUint(ticket, 10)
 	if _, err := fmt.Fprintf(w, "%d", ticket); err != nil {
 		log.Printf("Writing response failed: %v", err)
 	}
 }
 
-func setupRoutes() {
-	http.HandleFunc("/", pageHandler)
-	http.HandleFunc("/new", newHandler)
-	http.HandleFunc("/done", doneHandler)
-}
-
 func main() {
-	setupRoutes()
+	// Make a new Broker instance
+	broker = &Broker{
+		make(map[chan string]bool),
+		make(chan (chan string)),
+		make(chan (chan string)),
+		make(chan string),
+	}
+
+	// Start processing events
+	broker.Start()
 
 	pageTemplate = template.Must(template.New("index").Parse(indexTemplate))
 
+	http.HandleFunc("/", pageHandler)
+	http.HandleFunc("/new", newHandler)
+	http.HandleFunc("/done", doneHandler)
+	http.Handle("/events", broker)
+
 	log.Printf("Serving http on %d", 8080)
 	err := http.ListenAndServe(fmt.Sprintf(":%d", 8080), nil)
 	log.Panicf("Serving http failed with %s", err)