Server-Sent Events (SSE) with heartbeats and client cleanup

                  package realtime

import (
  "fmt"
  "net/http"
  "time"
)

type Hub struct {
  subscribe   chan chan string
  unsubscribe chan chan string
  broadcast   chan string
}

func NewHub() *Hub {
  h := &Hub{subscribe: make(chan chan string), unsubscribe: make(chan chan string), broadcast: make(chan string, 128)}
  go h.run()
  return h
}

func (h *Hub) run() {
  clients := map[chan string]struct{}{}
  for {
    select {
    case c := <-h.subscribe:
      clients[c] = struct{}{}
    case c := <-h.unsubscribe:
      delete(clients, c)
      close(c)
    case msg := <-h.broadcast:
      for c := range clients {
        select {
        case c <- msg:
        default:
        }
      }
    }
  }
}

func (h *Hub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  flusher, ok := w.(http.Flusher)
  if !ok {
    http.Error(w, "streaming unsupported", http.StatusInternalServerError)
    return
  }
  w.Header().Set("Content-Type", "text/event-stream")
  w.Header().Set("Cache-Control", "no-cache")
  w.Header().Set("Connection", "keep-alive")

  ch := make(chan string, 16)
  h.subscribe <- ch
  defer func() { h.unsubscribe <- ch }()

  heartbeat := time.NewTicker(15 * time.Second)
  defer heartbeat.Stop()

  for {
    select {
    case <-r.Context().Done():
      return
    case <-heartbeat.C:
      fmt.Fprint(w, ": ping

")
      flusher.Flush()
    case msg := <-ch:
      fmt.Fprintf(w, "data: %s

", msg)
      flusher.Flush()
    }
  }
}