Mikkel Georgsen
9f57cbdf6c
- LabelHandler: POST /api/labels/:deviceID/print with 1/s rate limit (T-04-09) - USBEventsHandler: GET /api/usb/events SSE stream, exits on context cancel (T-04-11) - router.go: two new parameters + routes wired - main.go: USB Manager started with ctx, MockDriver connected, handlers passed to router
85 lines
2.3 KiB
Go
85 lines
2.3 KiB
Go
package handlers
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
|
|
"git.georgsen.dk/hwlab/internal/usb"
|
|
)
|
|
|
|
// USBEventSource is the narrow interface over usb.Manager that the SSE handler needs.
|
|
// *usb.Manager satisfies this interface.
|
|
type USBEventSource interface {
|
|
Events() <-chan usb.DeviceEvent
|
|
}
|
|
|
|
// USBEventsHandler handles GET /api/usb/events — SSE stream of USB device events.
|
|
// T-04-11 mitigation: handler returns on r.Context().Done() — no goroutine leak.
|
|
type USBEventsHandler struct {
|
|
manager USBEventSource
|
|
}
|
|
|
|
// NewUSBEventsHandler creates a USBEventsHandler backed by the given event source.
|
|
func NewUSBEventsHandler(m USBEventSource) *USBEventsHandler {
|
|
return &USBEventsHandler{manager: m}
|
|
}
|
|
|
|
// ServeEvents streams USB device connect/disconnect events as Server-Sent Events.
|
|
//
|
|
// SSE format (per WHATWG):
|
|
//
|
|
// data: {"VIDPID":"0525:a4a7","Spec":{...},"State":1}\n\n
|
|
//
|
|
// A keepalive comment (": keepalive\n\n") is sent every 30 seconds to prevent
|
|
// proxy and load-balancer timeouts.
|
|
//
|
|
// The goroutine exits cleanly when the client disconnects (r.Context().Done()).
|
|
func (h *USBEventsHandler) ServeEvents(w http.ResponseWriter, r *http.Request) {
|
|
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") // disable nginx proxy buffering
|
|
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "streaming not supported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Initial keepalive confirms connection to the client.
|
|
fmt.Fprintf(w, ": connected\n\n")
|
|
flusher.Flush()
|
|
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
// Client disconnected — T-04-11: return cleanly, no goroutine leak.
|
|
return
|
|
|
|
case evt, ok := <-h.manager.Events():
|
|
if !ok {
|
|
// Channel closed (Manager stopped) — close SSE stream.
|
|
return
|
|
}
|
|
data, err := json.Marshal(evt)
|
|
if err != nil {
|
|
// Non-fatal: log and continue rather than killing the stream.
|
|
fmt.Fprintf(w, ": marshal error\n\n")
|
|
flusher.Flush()
|
|
continue
|
|
}
|
|
fmt.Fprintf(w, "data: %s\n\n", data)
|
|
flusher.Flush()
|
|
|
|
case <-ticker.C:
|
|
// Keepalive comment — prevents proxy timeout on quiet periods.
|
|
fmt.Fprintf(w, ": keepalive\n\n")
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|