From 970fffc0cc7a3ee898df8ee5838a512c52d96df7 Mon Sep 17 00:00:00 2001 From: Florian Fischer Date: Sun, 25 Jan 2026 10:15:29 +0100 Subject: distribute the next ticket via SSE --- index.html | 8 +++- main.go | 142 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--- 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; + }; 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) -- cgit v1.2.3