83 lines
2.3 KiB
Go
83 lines
2.3 KiB
Go
package events
|
|
|
|
import (
|
|
"io"
|
|
"log/slog"
|
|
"sync"
|
|
"testing"
|
|
)
|
|
|
|
func TestStoreBoundsCopiesAndConcurrentCounters(t *testing.T) {
|
|
s := NewStore(4, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 10; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for n := 0; n < 10; n++ {
|
|
s.Add(Event{Kind: MessageReceived, Properties: Properties{CorrelationData: []byte("id")}})
|
|
s.Snapshot()
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
history, counts := s.Snapshot()
|
|
if len(history) != 4 || counts[MessageReceived] != 100 {
|
|
t.Fatal("history/counter bound incorrect")
|
|
}
|
|
for i := 1; i < len(history); i++ {
|
|
if history[i].ID <= history[i-1].ID {
|
|
t.Fatal("history IDs out of order")
|
|
}
|
|
}
|
|
history[0].Properties.CorrelationData[0] = 'X'
|
|
again, _ := s.Snapshot()
|
|
if string(again[0].Properties.CorrelationData) != "id" {
|
|
t.Fatal("snapshot mutated internal history")
|
|
}
|
|
}
|
|
|
|
func TestSubscribeReplayGapLiveFanoutAndCleanup(t *testing.T) {
|
|
s := NewStore(3, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
for i := 0; i < 5; i++ {
|
|
s.Add(Event{Kind: MessageReceived})
|
|
}
|
|
replay, gap, stream, unsubscribe := s.Subscribe(0, 1, 1)
|
|
if gap == nil || gap.ResetID != 4 || gap.Gap != 4 || len(replay) != 1 || replay[0].ID != 5 {
|
|
t.Fatalf("bounded replay gap: %#v %#v", gap, replay)
|
|
}
|
|
s.Add(Event{Kind: MessageReceived})
|
|
if n := <-stream; n.Event == nil || n.Event.ID != 6 {
|
|
t.Fatal("live event fanout failed")
|
|
}
|
|
unsubscribe()
|
|
unsubscribe()
|
|
if len(s.subs) != 0 {
|
|
t.Fatal("subscriber cleanup failed")
|
|
}
|
|
_, gap, _, closeSub := s.Subscribe(999, 1, 1)
|
|
defer closeSub()
|
|
if gap == nil || gap.ResetID != 6 {
|
|
t.Fatal("future cursor did not request reset")
|
|
}
|
|
}
|
|
func TestSlowViewerGetsGapWithoutBlockingEvents(t *testing.T) {
|
|
s := NewStore(8, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
s.Add(Event{Kind: MessageReceived})
|
|
_, _, stream, unsubscribe := s.Subscribe(1, 1, 8)
|
|
defer unsubscribe()
|
|
s.Add(Event{Kind: MessageReceived})
|
|
s.Add(Event{Kind: MessageReceived})
|
|
if n := <-stream; n.Event == nil || n.Event.ID != 2 {
|
|
t.Fatal("first bounded event missing")
|
|
}
|
|
s.Add(Event{Kind: MessageReceived})
|
|
if n := <-stream; n.Gap != 2 || n.ResetID != 4 {
|
|
t.Fatalf("missing explicit overflow gap: %#v", n)
|
|
}
|
|
s.Add(Event{Kind: MessageReceived})
|
|
if n := <-stream; n.Event == nil || n.Event.ID != 5 {
|
|
t.Fatalf("unexpected event after overflow gap: %#v", n)
|
|
}
|
|
}
|