josh
9bb4b09a04
CI / Lint + build + test (push) Has been cancelled
Post-repair hardware validation pipeline for Proxmox cluster hosts. Go orchestrator + in-image agent + mkosi live image + bundled dnsmasq PXE + SQLite + HTMX/SSE UI + notify registry + janitor + full docs.
145 lines
3.0 KiB
Go
145 lines
3.0 KiB
Go
package events
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// Event is a typed event published on the internal bus. In Phase 1 the
|
|
// payload is an already-rendered HTML fragment; later phases will wrap
|
|
// structured run state in this same Event envelope.
|
|
type Event struct {
|
|
Name string // SSE event name (e.g. "heartbeat", "tile-update", "log-line")
|
|
Payload string // pre-rendered HTML, ready to write as SSE data
|
|
}
|
|
|
|
type subscriber struct {
|
|
id int64
|
|
ch chan Event
|
|
}
|
|
|
|
// Hub is an in-process fan-out for SSE subscribers.
|
|
type Hub struct {
|
|
mu sync.RWMutex
|
|
nextID int64
|
|
subs map[int64]*subscriber
|
|
buffer int
|
|
heartbeat time.Duration
|
|
}
|
|
|
|
func NewHub() *Hub {
|
|
h := &Hub{
|
|
subs: map[int64]*subscriber{},
|
|
buffer: 32,
|
|
heartbeat: 15 * time.Second,
|
|
}
|
|
go h.heartbeatLoop()
|
|
return h
|
|
}
|
|
|
|
func (h *Hub) Publish(ev Event) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
for _, s := range h.subs {
|
|
select {
|
|
case s.ch <- ev:
|
|
default:
|
|
// Slow subscriber: drop the event rather than stall other clients.
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *Hub) Subscribe() (id int64, ch <-chan Event, cancel func()) {
|
|
id = atomic.AddInt64(&h.nextID, 1)
|
|
s := &subscriber{id: id, ch: make(chan Event, h.buffer)}
|
|
h.mu.Lock()
|
|
h.subs[id] = s
|
|
h.mu.Unlock()
|
|
return id, s.ch, func() {
|
|
h.mu.Lock()
|
|
delete(h.subs, id)
|
|
h.mu.Unlock()
|
|
close(s.ch)
|
|
}
|
|
}
|
|
|
|
func (h *Hub) heartbeatLoop() {
|
|
t := time.NewTicker(h.heartbeat)
|
|
defer t.Stop()
|
|
for range t.C {
|
|
h.Publish(Event{
|
|
Name: "heartbeat",
|
|
Payload: fmt.Sprintf(`<span data-heartbeat="%d"></span>`, time.Now().Unix()),
|
|
})
|
|
}
|
|
}
|
|
|
|
// ServeSSE writes server-sent events for a single subscriber for the
|
|
// lifetime of the request. Each Event becomes one SSE message.
|
|
func (h *Hub) ServeSSE(w http.ResponseWriter, r *http.Request) {
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "streaming not supported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
|
|
_, eventsCh, cancel := h.Subscribe()
|
|
defer cancel()
|
|
|
|
fmt.Fprintf(w, "event: hello\ndata: ok\n\n")
|
|
flusher.Flush()
|
|
|
|
ctx := r.Context()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case ev, ok := <-eventsCh:
|
|
if !ok {
|
|
return
|
|
}
|
|
writeSSE(w, ev)
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
func writeSSE(w http.ResponseWriter, ev Event) {
|
|
if ev.Name != "" {
|
|
fmt.Fprintf(w, "event: %s\n", ev.Name)
|
|
}
|
|
for _, line := range splitLines(ev.Payload) {
|
|
fmt.Fprintf(w, "data: %s\n", line)
|
|
}
|
|
fmt.Fprint(w, "\n")
|
|
}
|
|
|
|
func splitLines(s string) []string {
|
|
if s == "" {
|
|
return []string{""}
|
|
}
|
|
out := []string{}
|
|
start := 0
|
|
for i := 0; i < len(s); i++ {
|
|
if s[i] == '\n' {
|
|
out = append(out, s[start:i])
|
|
start = i + 1
|
|
}
|
|
}
|
|
if start <= len(s) {
|
|
out = append(out, s[start:])
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Shutdown is a no-op placeholder wired into graceful shutdown.
|
|
func (h *Hub) Shutdown(_ context.Context) error { return nil }
|