package main
import (
_ "embed"
"fmt"
"html/template"
"log"
"net/http"
"strconv"
"sync/atomic"
)
//go:embed index.html
var indexTemplate string
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)
return
}
tempData := struct {
Open uint64
Done uint64
IsAdmin bool
}{open.Load(), done.Load() + 1, r.URL.Query().Has("admin")}
if err := pageTemplate.Execute(w, tempData); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
func newHandler(w http.ResponseWriter, r *http.Request) {
ticket := open.Add(1)
log.Printf("Create ticket %d\n", ticket)
if _, err := fmt.Fprintf(w, "%d", ticket); err != nil {
log.Printf("Writing response failed: %v", err)
}
}
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 main() {
// 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)
}