rc-1

backend/tools/http_events.go

@@ -0,0 +1,82 @@
+package tools
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+)
+
+var (
+	ErrEventsUnsupported = errors.New("events unsupported")
+)
+
+type EventHelper struct {
+	m sync.Mutex
+	c context.Context
+	w http.ResponseWriter
+	r *http.Request
+	f http.Flusher
+}
+
+func NewEventHelper(ctx context.Context, w http.ResponseWriter, r *http.Request) (*EventHelper, error) {
+
+	// Setup Connection
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		return nil, ErrEventsUnsupported
+	}
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	sse := &EventHelper{f: flusher, c: ctx, w: w, r: r}
+
+	// Heartbeat Generator
+	go func(h *EventHelper) {
+		t := time.NewTicker(5 * time.Second)
+		defer t.Stop()
+		for {
+			select {
+			case <-h.c.Done():
+				return
+			case <-t.C:
+				h.m.Lock()
+				fmt.Fprintf(h.w, ": ping\n\n")
+				h.f.Flush()
+				h.m.Unlock()
+			}
+		}
+	}(sse)
+
+	return sse, nil
+}
+
+func (h *EventHelper) SendJSON(eventName string, eventData any) {
+
+	b, err := json.Marshal(map[string]any{
+		"name": eventName,
+		"data": eventData,
+	})
+	if err != nil {
+		return
+	}
+
+	h.m.Lock()
+	fmt.Fprintf(h.w, "data: %s\n\n", b)
+	h.f.Flush()
+	h.m.Unlock()
+}
+
+func (h *EventHelper) SendServerError(err error) {
+	SendServerError(nil, h.r, err)
+	h.SendClientError(ERROR_GENERIC_SERVER)
+}
+
+func (h *EventHelper) SendClientError(err APIError) {
+	h.SendJSON("error", err)
+}