dtoro
7e802bbb14
The generated strict-server path could only return an io.Reader that io.Copy drains without flushing, so SSE events sat chunk-buffered instead of streaming in real time. Replace it with a raw http.ResponseWriter handler (serveSSE) that Flush()es after every event. Routing: chi allows a later registration to supersede an earlier one for the same method+path (verified empirically for v5.3.1), so serveSSE is registered on the router AFTER gen.HandlerWithOptions and wins over the generated /events/stream route. It inherits the base middleware chain and applies auth via With(). The generated StreamEvents method now returns an error (never reached) so a routing regression fails loudly rather than silently reverting to buffered delivery. Adds TestSSEStreamRealtimeDelivery: a real httptest.NewServer + streaming client (NewRecorder can't flush) that connects, triggers an event, and asserts delivery within 3s — proving both the override routing and per-event flushing. Passes in <1s. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
90 lines
2.8 KiB
Go
90 lines
2.8 KiB
Go
package httpapi
|
|
|
|
// Real-connection SSE test. httptest.NewRecorder buffers and never flushes,
|
|
// so this uses httptest.NewServer + a streaming client to verify that:
|
|
// - the raw serveSSE handler (not the generated 501 stub) serves the route,
|
|
// - an event created *after* the client connects is delivered in real time
|
|
// (i.e. flushed before the connection closes),
|
|
// - the SSE `data:` payload is the canonical gen.Event shape.
|
|
// Guarded by OIKOS_TEST_DATABASE_URL.
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestSSEStreamRealtimeDelivery(t *testing.T) {
|
|
srv := httptest.NewServer(newTestHandler(t, devConfig()))
|
|
defer srv.Close()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
// Connect to the stream.
|
|
req, _ := http.NewRequestWithContext(ctx, "GET", srv.URL+"/api/v1/events/stream", nil)
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
t.Fatalf("connect stream: %v", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != 200 {
|
|
t.Fatalf("stream status = %d, want 200", resp.StatusCode)
|
|
}
|
|
if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
|
|
t.Fatalf("content-type = %q, want text/event-stream", ct)
|
|
}
|
|
|
|
// Read SSE frames in a goroutine.
|
|
dataCh := make(chan map[string]any, 4)
|
|
go func() {
|
|
sc := bufio.NewScanner(resp.Body)
|
|
for sc.Scan() {
|
|
line := sc.Text()
|
|
if strings.HasPrefix(line, "data: ") {
|
|
var m map[string]any
|
|
if json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &m) == nil {
|
|
dataCh <- m
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
|
|
// Give the subscriber a moment to register, then trigger an event by
|
|
// POSTing to the SAME live server (same DB → NOTIFY the listener sees).
|
|
time.Sleep(300 * time.Millisecond)
|
|
payload, _ := json.Marshal(map[string]any{"slug": "service:sse-rt", "type": "service", "name": "sse-rt"})
|
|
cResp, err := http.Post(srv.URL+"/api/v1/entities", "application/json", bytes.NewReader(payload))
|
|
if err != nil {
|
|
t.Fatalf("trigger create: %v", err)
|
|
}
|
|
cResp.Body.Close()
|
|
if cResp.StatusCode != 201 {
|
|
t.Fatalf("trigger create status = %d, want 201", cResp.StatusCode)
|
|
}
|
|
|
|
// The event must arrive in real time (well before the 10s ctx deadline),
|
|
// proving the handler flushes rather than buffering until close.
|
|
select {
|
|
case ev := <-dataCh:
|
|
if ev["type"] != "entity.created" {
|
|
t.Errorf("event type = %v, want entity.created", ev["type"])
|
|
}
|
|
// canonical shape: snake_case + decoded data object
|
|
if _, ok := ev["entity_id"]; !ok {
|
|
t.Errorf("missing snake_case entity_id: %v", ev)
|
|
}
|
|
if d, ok := ev["data"].(map[string]any); !ok || d["slug"] != "service:sse-rt" {
|
|
t.Errorf("data not a decoded object with slug: %v", ev["data"])
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("SSE event not delivered within 3s (flushing broken?)")
|
|
}
|
|
}
|