420 lines
13 KiB
Go
420 lines
13 KiB
Go
package stream
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"reflect"
|
||
"strings"
|
||
"sync"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
// eventually wartet, bis cond wahr ist, höchstens zwei Sekunden.
|
||
func eventually(t *testing.T, msg string, cond func() bool) {
|
||
t.Helper()
|
||
deadline := time.Now().Add(2 * time.Second)
|
||
for !cond() {
|
||
if time.Now().After(deadline) {
|
||
t.Fatalf("Bedingung nicht erfüllt: %s", msg)
|
||
}
|
||
time.Sleep(2 * time.Millisecond)
|
||
}
|
||
}
|
||
|
||
// hubServer startet einen Testserver, der Verbindungen an h übergibt. Die
|
||
// Knoten-ID kommt aus ?node=. Rückgabewerte von Serve landen in errs.
|
||
func hubServer(t *testing.T, h *Hub) (srv *httptest.Server, errs chan error) {
|
||
t.Helper()
|
||
errs = make(chan error, 16)
|
||
srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
errs <- h.Serve(w, r, r.URL.Query().Get("node"))
|
||
}))
|
||
t.Cleanup(srv.Close)
|
||
t.Cleanup(h.Close)
|
||
return srv, errs
|
||
}
|
||
|
||
func openStream(t *testing.T, ctx context.Context, srv *httptest.Server, node string) *http.Response {
|
||
t.Helper()
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"?node="+node, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
resp, err := srv.Client().Do(req)
|
||
if err != nil {
|
||
t.Fatalf("Verbindung fehlgeschlagen: %v", err)
|
||
}
|
||
t.Cleanup(func() { resp.Body.Close() })
|
||
return resp
|
||
}
|
||
|
||
func waitErr(t *testing.T, errs <-chan error) error {
|
||
t.Helper()
|
||
select {
|
||
case err := <-errs:
|
||
return err
|
||
case <-time.After(2 * time.Second):
|
||
t.Fatal("Serve ist nicht zurückgekehrt")
|
||
return nil
|
||
}
|
||
}
|
||
|
||
func TestHub_SendOffline(t *testing.T) {
|
||
var h Hub
|
||
if err := h.Send("node-1", Event{Type: EventNodeCommand}); !errors.Is(err, ErrNodeOffline) {
|
||
t.Errorf("erwartet ErrNodeOffline, bekommen %v", err)
|
||
}
|
||
}
|
||
|
||
func TestHub_Defaults(t *testing.T) {
|
||
var h Hub
|
||
if got := h.keepAlive(); got != 45*time.Second {
|
||
t.Errorf("KeepAlive-Default = %v, erwartet 45s (Spec)", got)
|
||
}
|
||
if got := h.writeTimeout(); got != 10*time.Second {
|
||
t.Errorf("WriteTimeout-Default = %v", got)
|
||
}
|
||
if got := h.bufferSize(); got != 16 {
|
||
t.Errorf("BufferSize-Default = %d", got)
|
||
}
|
||
}
|
||
|
||
func TestHub_DeliversEvent(t *testing.T) {
|
||
h := &Hub{}
|
||
srv, _ := hubServer(t, h)
|
||
resp := openStream(t, context.Background(), srv, "node-1")
|
||
|
||
if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" {
|
||
t.Errorf("Content-Type = %q", ct)
|
||
}
|
||
if cc := resp.Header.Get("Cache-Control"); cc != "no-cache" {
|
||
t.Errorf("Cache-Control = %q", cc)
|
||
}
|
||
eventually(t, "node-1 verbunden", func() bool { return h.Connected("node-1") })
|
||
|
||
if err := h.Send("node-1", Event{Type: EventNodeCommand, Payload: json.RawMessage(`{"action":"drain"}`)}); err != nil {
|
||
t.Fatalf("Send: %v", err)
|
||
}
|
||
ev, err := NewDecoder(resp.Body).Next()
|
||
if err != nil {
|
||
t.Fatalf("Next: %v", err)
|
||
}
|
||
if ev.Type != EventNodeCommand || ev.NodeID != "node-1" || string(ev.Payload) != `{"action":"drain"}` {
|
||
t.Errorf("unerwartetes Event: %+v", ev)
|
||
}
|
||
if !strings.HasPrefix(ev.ID, "evt_") || ev.Timestamp == 0 {
|
||
t.Errorf("ID/Timestamp nicht ergänzt: %+v", ev)
|
||
}
|
||
}
|
||
|
||
func TestHub_SendKeepsIDAndTimestamp(t *testing.T) {
|
||
h := &Hub{}
|
||
srv, _ := hubServer(t, h)
|
||
resp := openStream(t, context.Background(), srv, "n")
|
||
eventually(t, "verbunden", func() bool { return h.Connected("n") })
|
||
|
||
want := Event{ID: "evt_fix", Timestamp: 42, Type: EventLeaseRenew, NodeID: "n"}
|
||
if err := h.Send("n", Event{ID: "evt_fix", Timestamp: 42, Type: EventLeaseRenew, NodeID: "anderer"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
got, err := NewDecoder(resp.Body).Next()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !reflect.DeepEqual(got, want) {
|
||
t.Errorf("got %+v, erwartet %+v (NodeID wird immer vom Hub gesetzt)", got, want)
|
||
}
|
||
}
|
||
|
||
func TestHub_SendInvalid(t *testing.T) {
|
||
h := &Hub{}
|
||
srv, _ := hubServer(t, h)
|
||
openStream(t, context.Background(), srv, "n")
|
||
eventually(t, "verbunden", func() bool { return h.Connected("n") })
|
||
|
||
if err := h.Send("n", Event{}); !errors.Is(err, ErrInvalidEvent) {
|
||
t.Errorf("leerer Typ: erwartet ErrInvalidEvent, bekommen %v", err)
|
||
}
|
||
if err := h.Send("n", Event{Type: EventError, Payload: json.RawMessage(`{`)}); !errors.Is(err, ErrInvalidEvent) {
|
||
t.Errorf("kaputte Payload: erwartet ErrInvalidEvent, bekommen %v", err)
|
||
}
|
||
}
|
||
|
||
func TestHub_KeepAlive(t *testing.T) {
|
||
h := &Hub{KeepAlive: 10 * time.Millisecond}
|
||
srv, _ := hubServer(t, h)
|
||
resp := openStream(t, context.Background(), srv, "n")
|
||
|
||
buf := make([]byte, 6)
|
||
if _, err := io.ReadFull(resp.Body, buf); err != nil {
|
||
t.Fatalf("lesen: %v", err)
|
||
}
|
||
if string(buf) != ":\n\n:\n\n" {
|
||
t.Errorf("erwartet zwei Keepalives \":\\n\\n\", bekommen %q", buf)
|
||
}
|
||
}
|
||
|
||
func TestHub_DisconnectDetected(t *testing.T) {
|
||
var mu sync.Mutex
|
||
var connects, disconnects []string
|
||
h := &Hub{
|
||
OnConnect: func(n string) { mu.Lock(); connects = append(connects, n); mu.Unlock() },
|
||
OnDisconnect: func(n string) { mu.Lock(); disconnects = append(disconnects, n); mu.Unlock() },
|
||
}
|
||
srv, errs := hubServer(t, h)
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
openStream(t, ctx, srv, "node-1")
|
||
eventually(t, "verbunden", func() bool { return h.Connected("node-1") })
|
||
|
||
cancel()
|
||
if err := waitErr(t, errs); !errors.Is(err, context.Canceled) {
|
||
t.Errorf("Serve: erwartet context.Canceled, bekommen %v", err)
|
||
}
|
||
if h.Connected("node-1") {
|
||
t.Error("node-1 noch verbunden")
|
||
}
|
||
if err := h.Send("node-1", Event{Type: EventError}); !errors.Is(err, ErrNodeOffline) {
|
||
t.Errorf("erwartet ErrNodeOffline, bekommen %v", err)
|
||
}
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
if !reflect.DeepEqual(connects, []string{"node-1"}) || !reflect.DeepEqual(disconnects, []string{"node-1"}) {
|
||
t.Errorf("Callbacks: connects=%v disconnects=%v", connects, disconnects)
|
||
}
|
||
}
|
||
|
||
func TestHub_ReconnectReplacesOld(t *testing.T) {
|
||
var connects, disconnects atomic.Int32
|
||
h := &Hub{
|
||
OnConnect: func(string) { connects.Add(1) },
|
||
OnDisconnect: func(string) { disconnects.Add(1) },
|
||
}
|
||
srv, errs := hubServer(t, h)
|
||
old := openStream(t, context.Background(), srv, "n")
|
||
eventually(t, "erste Verbindung", func() bool { return connects.Load() == 1 })
|
||
fresh := openStream(t, context.Background(), srv, "n")
|
||
|
||
if err := waitErr(t, errs); !errors.Is(err, ErrReplaced) {
|
||
t.Errorf("alte Verbindung: erwartet ErrReplaced, bekommen %v", err)
|
||
}
|
||
if _, err := io.ReadAll(old.Body); err != nil {
|
||
t.Errorf("alter Stream sollte sauber enden: %v", err)
|
||
}
|
||
if err := h.Send("n", Event{Type: EventLeaseRenew}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ev, err := NewDecoder(fresh.Body).Next()
|
||
if err != nil || ev.Type != EventLeaseRenew {
|
||
t.Errorf("neue Verbindung: ev=%+v err=%v", ev, err)
|
||
}
|
||
if c, d := connects.Load(), disconnects.Load(); c != 2 || d != 0 {
|
||
t.Errorf("Callbacks: connects=%d disconnects=%d, erwartet 2/0 (Ersatz ist kein Offline)", c, d)
|
||
}
|
||
}
|
||
|
||
func TestHub_Nodes(t *testing.T) {
|
||
h := &Hub{}
|
||
srv, _ := hubServer(t, h)
|
||
openStream(t, context.Background(), srv, "b")
|
||
openStream(t, context.Background(), srv, "a")
|
||
eventually(t, "beide verbunden", func() bool { return len(h.Nodes()) == 2 })
|
||
if got := h.Nodes(); !reflect.DeepEqual(got, []string{"a", "b"}) {
|
||
t.Errorf("Nodes = %v, erwartet sortiert [a b]", got)
|
||
}
|
||
}
|
||
|
||
func TestHub_Close(t *testing.T) {
|
||
h := &Hub{}
|
||
srv, errs := hubServer(t, h)
|
||
resp := openStream(t, context.Background(), srv, "n")
|
||
eventually(t, "verbunden", func() bool { return h.Connected("n") })
|
||
|
||
h.Close()
|
||
if err := waitErr(t, errs); !errors.Is(err, ErrHubClosed) {
|
||
t.Errorf("Serve: erwartet ErrHubClosed, bekommen %v", err)
|
||
}
|
||
if _, err := io.ReadAll(resp.Body); err != nil {
|
||
t.Errorf("Stream sollte sauber enden: %v", err)
|
||
}
|
||
if err := h.Send("n", Event{Type: EventError}); !errors.Is(err, ErrHubClosed) {
|
||
t.Errorf("Send nach Close: erwartet ErrHubClosed, bekommen %v", err)
|
||
}
|
||
req := httptest.NewRequest(http.MethodGet, "/", nil)
|
||
if err := h.Serve(httptest.NewRecorder(), req, "neu"); !errors.Is(err, ErrHubClosed) {
|
||
t.Errorf("Serve nach Close: erwartet ErrHubClosed, bekommen %v", err)
|
||
}
|
||
h.Close() // zweimal schließen darf nicht paniken
|
||
}
|
||
|
||
// blockingWriter simuliert einen Client, der nicht mehr liest.
|
||
type blockingWriter struct {
|
||
header http.Header
|
||
release chan struct{}
|
||
}
|
||
|
||
func (w *blockingWriter) Header() http.Header { return w.header }
|
||
func (w *blockingWriter) WriteHeader(int) {}
|
||
func (w *blockingWriter) Flush() {}
|
||
func (w *blockingWriter) Write(p []byte) (int, error) { <-w.release; return len(p), nil }
|
||
|
||
func TestHub_BufferFull(t *testing.T) {
|
||
h := &Hub{BufferSize: 2}
|
||
w := &blockingWriter{header: http.Header{}, release: make(chan struct{})}
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
done := make(chan error, 1)
|
||
go func() { done <- h.Serve(w, httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx), "n") }()
|
||
eventually(t, "verbunden", func() bool { return h.Connected("n") })
|
||
|
||
var full bool
|
||
for range 4 {
|
||
if err := h.Send("n", Event{Type: EventError}); errors.Is(err, ErrBufferFull) {
|
||
full = true
|
||
}
|
||
}
|
||
if !full {
|
||
t.Error("erwartet ErrBufferFull bei vollem Puffer")
|
||
}
|
||
cancel()
|
||
close(w.release)
|
||
if err := waitErr(t, done); !errors.Is(err, context.Canceled) {
|
||
t.Errorf("Serve: erwartet context.Canceled, bekommen %v", err)
|
||
}
|
||
}
|
||
|
||
type failingWriter struct{ header http.Header }
|
||
|
||
var errWrite = errors.New("Verbindung weg")
|
||
|
||
func (w *failingWriter) Header() http.Header { return w.header }
|
||
func (w *failingWriter) WriteHeader(int) {}
|
||
func (w *failingWriter) Flush() {}
|
||
func (w *failingWriter) Write([]byte) (int, error) { return 0, errWrite }
|
||
|
||
func TestHub_WriteErrorEndsServe(t *testing.T) {
|
||
h := &Hub{KeepAlive: 5 * time.Millisecond}
|
||
err := h.Serve(&failingWriter{header: http.Header{}}, httptest.NewRequest(http.MethodGet, "/", nil), "n")
|
||
if !errors.Is(err, errWrite) {
|
||
t.Errorf("erwartet Schreibfehler, bekommen %v", err)
|
||
}
|
||
if h.Connected("n") {
|
||
t.Error("Knoten nach Schreibfehler noch verbunden")
|
||
}
|
||
}
|
||
|
||
type noFlushWriter struct{ header http.Header }
|
||
|
||
func (w *noFlushWriter) Header() http.Header { return w.header }
|
||
func (w *noFlushWriter) WriteHeader(int) {}
|
||
func (w *noFlushWriter) Write(p []byte) (int, error) { return len(p), nil }
|
||
|
||
func TestHub_StreamingUnsupported(t *testing.T) {
|
||
var h Hub
|
||
err := h.Serve(&noFlushWriter{header: http.Header{}}, httptest.NewRequest(http.MethodGet, "/", nil), "n")
|
||
if !errors.Is(err, ErrStreamingUnsupported) {
|
||
t.Errorf("erwartet ErrStreamingUnsupported, bekommen %v", err)
|
||
}
|
||
if h.Connected("n") {
|
||
t.Error("Knoten darf nicht registriert sein")
|
||
}
|
||
}
|
||
|
||
func TestHub_ConcurrentSend(t *testing.T) {
|
||
const senders, perSender = 20, 50
|
||
h := &Hub{}
|
||
srv, _ := hubServer(t, h)
|
||
resp := openStream(t, context.Background(), srv, "n")
|
||
eventually(t, "verbunden", func() bool { return h.Connected("n") })
|
||
|
||
var wg sync.WaitGroup
|
||
for range senders {
|
||
wg.Go(func() {
|
||
for range perSender {
|
||
for {
|
||
err := h.Send("n", Event{Type: EventServiceState})
|
||
if err == nil {
|
||
break
|
||
}
|
||
if !errors.Is(err, ErrBufferFull) {
|
||
t.Errorf("Send: %v", err)
|
||
return
|
||
}
|
||
time.Sleep(time.Millisecond)
|
||
}
|
||
}
|
||
})
|
||
}
|
||
dec := NewDecoder(resp.Body)
|
||
for i := range senders * perSender {
|
||
if _, err := dec.Next(); err != nil {
|
||
t.Fatalf("Event %d: %v", i, err)
|
||
}
|
||
}
|
||
wg.Wait()
|
||
}
|
||
|
||
func TestHub_ServeRejectsEmptyNodeID(t *testing.T) {
|
||
var connects atomic.Int32
|
||
h := &Hub{OnConnect: func(string) { connects.Add(1) }}
|
||
rec := httptest.NewRecorder()
|
||
err := h.Serve(rec, httptest.NewRequest(http.MethodGet, "/", nil), "")
|
||
if !errors.Is(err, ErrInvalidNodeID) {
|
||
t.Errorf("erwartet ErrInvalidNodeID, bekommen %v", err)
|
||
}
|
||
if rec.Body.Len() != 0 || rec.Header().Get("Content-Type") == "text/event-stream" {
|
||
t.Error("bei ungültiger ID darf kein Stream begonnen werden")
|
||
}
|
||
if connects.Load() != 0 || len(h.Nodes()) != 0 {
|
||
t.Error("leere ID wurde registriert")
|
||
}
|
||
}
|
||
|
||
func TestHub_Disconnect(t *testing.T) {
|
||
var disconnects atomic.Int32
|
||
h := &Hub{OnDisconnect: func(string) { disconnects.Add(1) }}
|
||
srv, errs := hubServer(t, h)
|
||
resp := openStream(t, context.Background(), srv, "n")
|
||
eventually(t, "verbunden", func() bool { return h.Connected("n") })
|
||
|
||
if err := h.Disconnect("n"); err != nil {
|
||
t.Fatalf("Disconnect: %v", err)
|
||
}
|
||
// Sofort wirksam, nicht erst wenn Serve zurückkehrt:
|
||
if h.Connected("n") {
|
||
t.Error("Knoten nach Disconnect noch verbunden")
|
||
}
|
||
if err := h.Send("n", Event{Type: EventError}); !errors.Is(err, ErrNodeOffline) {
|
||
t.Errorf("Send nach Disconnect: erwartet ErrNodeOffline, bekommen %v", err)
|
||
}
|
||
if err := waitErr(t, errs); !errors.Is(err, ErrDisconnected) {
|
||
t.Errorf("Serve: erwartet ErrDisconnected, bekommen %v", err)
|
||
}
|
||
if _, err := io.ReadAll(resp.Body); err != nil {
|
||
t.Errorf("Stream sollte sauber enden: %v", err)
|
||
}
|
||
if n := disconnects.Load(); n != 1 {
|
||
t.Errorf("OnDisconnect %d-mal aufgerufen, erwartet 1", n)
|
||
}
|
||
if err := h.Disconnect("n"); !errors.Is(err, ErrNodeOffline) {
|
||
t.Errorf("zweites Disconnect: erwartet ErrNodeOffline, bekommen %v", err)
|
||
}
|
||
}
|
||
|
||
// Sobald der Client die Antwort-Header hat, muss der Knoten registriert sein –
|
||
// sonst scheitert ein direkt folgender Request ("Stream nicht verbunden").
|
||
func TestHub_ConnectedWhenHeadersArrive(t *testing.T) {
|
||
for range 50 {
|
||
h := &Hub{}
|
||
srv, _ := hubServer(t, h)
|
||
openStream(t, context.Background(), srv, "n")
|
||
if !h.Connected("n") {
|
||
t.Fatal("Header empfangen, Knoten aber noch nicht registriert")
|
||
}
|
||
}
|
||
}
|