panel v0.9.1 — open-source game server manager
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,228 @@
|
||||
// Package events is the Controller's in-memory pub-sub for live agent
|
||||
// events (instance state, app state, player, log). The Panel service's
|
||||
// StreamEvents RPC subscribes; agent message handlers publish. Subscribers
|
||||
// have bounded buffers and are dropped rather than back-pressuring the
|
||||
// whole pipeline if they can't keep up.
|
||||
package events
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
panelv1 "github.com/dbledeez/panel/proto/panel/v1"
|
||||
)
|
||||
|
||||
// Filter narrows which events a subscriber receives. Empty string on a
|
||||
// field means "no filter on that field".
|
||||
type Filter struct {
|
||||
InstanceID string
|
||||
AgentID string
|
||||
// ExcludeLogs drops Event_Log payloads for this subscriber. The global
|
||||
// dashboard stream sets this: log lines are by far the bulk of event
|
||||
// volume (a 7DTD boot is thousands of lines in seconds), the dashboard
|
||||
// has no DOM target for them, and shipping them anyway is what
|
||||
// overflowed the subscriber buffer (dropped events → stale status) and
|
||||
// kept the browser main thread busy. Consoles + readiness scanners use
|
||||
// per-instance subscriptions, which keep their logs.
|
||||
ExcludeLogs bool
|
||||
}
|
||||
|
||||
// Subscription is the subscriber's handle. Call Close when done.
|
||||
type Subscription struct {
|
||||
bus *Bus
|
||||
id int64
|
||||
filter Filter
|
||||
ch chan *panelv1.Event
|
||||
// sch is non-nil only for SubscribeStamped subscribers; exactly one of
|
||||
// ch/sch carries events for a given subscription.
|
||||
sch chan Stamped
|
||||
dropped atomic.Uint64
|
||||
closed atomic.Bool
|
||||
}
|
||||
|
||||
// Events returns the receive channel. Consumer should range over it.
|
||||
func (s *Subscription) Events() <-chan *panelv1.Event { return s.ch }
|
||||
|
||||
// Stamped returns the id-stamped receive channel. Only valid on
|
||||
// subscriptions created via SubscribeStamped (nil otherwise).
|
||||
func (s *Subscription) Stamped() <-chan Stamped { return s.sch }
|
||||
|
||||
// Dropped returns the number of events that were dropped because the
|
||||
// subscriber's buffer was full.
|
||||
func (s *Subscription) Dropped() uint64 { return s.dropped.Load() }
|
||||
|
||||
// Close unsubscribes and closes the channel. Safe to call multiple times.
|
||||
func (s *Subscription) Close() {
|
||||
if !s.closed.CompareAndSwap(false, true) {
|
||||
return
|
||||
}
|
||||
s.bus.unsubscribe(s.id)
|
||||
if s.sch != nil {
|
||||
close(s.sch)
|
||||
} else {
|
||||
close(s.ch)
|
||||
}
|
||||
}
|
||||
|
||||
// Bus is a concurrency-safe event fan-out with per-subscriber bounded
|
||||
// buffers. Subscribers register with optional filters; publishers deliver
|
||||
// each matching event non-blocking (drop on full buffer).
|
||||
type Bus struct {
|
||||
mu sync.RWMutex
|
||||
next int64
|
||||
subs map[int64]*Subscription
|
||||
bufCap int
|
||||
// ring retains the most recent events (with monotonic ids) so SSE
|
||||
// clients can resume via Last-Event-ID. See ring.go.
|
||||
ring *replayRing
|
||||
// droppedTotal counts every event dropped on a full subscriber buffer,
|
||||
// across all subscribers (including since-closed ones). Metrics surface.
|
||||
droppedTotal atomic.Uint64
|
||||
}
|
||||
|
||||
// New creates a Bus. bufCap is the per-subscriber channel buffer.
|
||||
//
|
||||
// The default is intentionally generous: a busy install/download (SteamCMD
|
||||
// preallocation, large Minecraft world seed, etc.) can emit 40+ log lines
|
||||
// per second per instance. At the old default of 256, a browser subscriber
|
||||
// that paused for half a second could lose dozens of intermediate events —
|
||||
// the user saw "0 → 88% → 7.11%" jumps because the progress-bar frames
|
||||
// between them hit the default: branch in Publish.
|
||||
func New(bufCap int) *Bus {
|
||||
if bufCap <= 0 {
|
||||
bufCap = 4096
|
||||
}
|
||||
return &Bus{subs: map[int64]*Subscription{}, bufCap: bufCap, ring: newReplayRing(DefaultRingCap)}
|
||||
}
|
||||
|
||||
// Subscribe returns a new Subscription. Call Sub.Close when done — typically
|
||||
// via defer from the stream handler.
|
||||
func (b *Bus) Subscribe(filter Filter) *Subscription {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.next++
|
||||
s := &Subscription{
|
||||
bus: b,
|
||||
id: b.next,
|
||||
filter: filter,
|
||||
ch: make(chan *panelv1.Event, b.bufCap),
|
||||
}
|
||||
b.subs[s.id] = s
|
||||
return s
|
||||
}
|
||||
|
||||
// SubscribeStamped is Subscribe, but the subscription's channel carries the
|
||||
// event's monotonic replay id alongside the event (Stamped()). Used by the
|
||||
// SSE handler to emit `id:` lines for Last-Event-ID resume.
|
||||
func (b *Bus) SubscribeStamped(filter Filter) *Subscription {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.next++
|
||||
s := &Subscription{
|
||||
bus: b,
|
||||
id: b.next,
|
||||
filter: filter,
|
||||
sch: make(chan Stamped, b.bufCap),
|
||||
}
|
||||
b.subs[s.id] = s
|
||||
return s
|
||||
}
|
||||
|
||||
// Replay returns the retained events with ID > afterID that match the
|
||||
// filter, oldest first. ok=false signals the resume point was overwritten
|
||||
// (a gap precedes the returned events) — callers should resync via snapshot.
|
||||
func (b *Bus) Replay(afterID uint64, f Filter) ([]Stamped, bool) {
|
||||
all, ok := b.ring.since(afterID)
|
||||
out := all[:0]
|
||||
for _, s := range all {
|
||||
if match(f, s.Event) {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out, ok
|
||||
}
|
||||
|
||||
// LastID returns the id of the most recently published event (0 if none).
|
||||
func (b *Bus) LastID() uint64 { return b.ring.lastID() }
|
||||
|
||||
// DroppedTotal returns the process-lifetime count of events dropped on
|
||||
// full subscriber buffers.
|
||||
func (b *Bus) DroppedTotal() uint64 { return b.droppedTotal.Load() }
|
||||
|
||||
func (b *Bus) unsubscribe(id int64) {
|
||||
b.mu.Lock()
|
||||
delete(b.subs, id)
|
||||
b.mu.Unlock()
|
||||
}
|
||||
|
||||
// Publish fans the event to every matching subscriber. Non-blocking —
|
||||
// slow subscribers lose events rather than back-pressuring publishers.
|
||||
func (b *Bus) Publish(e *panelv1.Event) {
|
||||
// Stamp + retain first so replay ids are assigned in publish order.
|
||||
id := b.ring.append(e)
|
||||
b.mu.RLock()
|
||||
subs := make([]*Subscription, 0, len(b.subs))
|
||||
for _, s := range b.subs {
|
||||
if match(s.filter, e) {
|
||||
subs = append(subs, s)
|
||||
}
|
||||
}
|
||||
b.mu.RUnlock()
|
||||
for _, s := range subs {
|
||||
if s.sch != nil {
|
||||
select {
|
||||
case s.sch <- Stamped{ID: id, Event: e}:
|
||||
default:
|
||||
s.dropped.Add(1)
|
||||
b.droppedTotal.Add(1)
|
||||
}
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case s.ch <- e:
|
||||
default:
|
||||
s.dropped.Add(1)
|
||||
b.droppedTotal.Add(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// SubscriberCount is useful for metrics / debug.
|
||||
func (b *Bus) SubscriberCount() int {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
return len(b.subs)
|
||||
}
|
||||
|
||||
func match(f Filter, e *panelv1.Event) bool {
|
||||
if f.AgentID != "" && f.AgentID != e.AgentId {
|
||||
return false
|
||||
}
|
||||
if f.InstanceID != "" {
|
||||
if eventInstanceID(e) != f.InstanceID {
|
||||
return false
|
||||
}
|
||||
}
|
||||
if f.ExcludeLogs {
|
||||
if _, isLog := e.Payload.(*panelv1.Event_Log); isLog {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func eventInstanceID(e *panelv1.Event) string {
|
||||
switch p := e.Payload.(type) {
|
||||
case *panelv1.Event_InstanceState:
|
||||
return p.InstanceState.InstanceId
|
||||
case *panelv1.Event_AppState:
|
||||
return p.AppState.InstanceId
|
||||
case *panelv1.Event_Player:
|
||||
return p.Player.InstanceId
|
||||
case *panelv1.Event_Log:
|
||||
return p.Log.InstanceId
|
||||
case *panelv1.Event_InstanceStats:
|
||||
return p.InstanceStats.InstanceId
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
panelv1 "github.com/dbledeez/panel/proto/panel/v1"
|
||||
)
|
||||
|
||||
func logEvent(agentID, instanceID string) *panelv1.Event {
|
||||
return &panelv1.Event{
|
||||
AgentId: agentID,
|
||||
Payload: &panelv1.Event_Log{Log: &panelv1.LogLine{InstanceId: instanceID, Line: "x"}},
|
||||
}
|
||||
}
|
||||
|
||||
func statsEvent(agentID, instanceID string) *panelv1.Event {
|
||||
return &panelv1.Event{
|
||||
AgentId: agentID,
|
||||
Payload: &panelv1.Event_InstanceStats{InstanceStats: &panelv1.InstanceStatsUpdate{InstanceId: instanceID}},
|
||||
}
|
||||
}
|
||||
|
||||
// drain returns however many events are immediately buffered.
|
||||
func drain(s *Subscription) int {
|
||||
n := 0
|
||||
for {
|
||||
select {
|
||||
case <-s.Events():
|
||||
n++
|
||||
default:
|
||||
return n
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilterExcludeLogs(t *testing.T) {
|
||||
b := New(16)
|
||||
sub := b.Subscribe(Filter{ExcludeLogs: true})
|
||||
defer sub.Close()
|
||||
|
||||
b.Publish(logEvent("a1", "i1"))
|
||||
b.Publish(statsEvent("a1", "i1"))
|
||||
b.Publish(logEvent("a2", "i2"))
|
||||
|
||||
if got := drain(sub); got != 1 {
|
||||
t.Fatalf("ExcludeLogs subscriber got %d events, want 1 (stats only)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilterInstanceAndAgent(t *testing.T) {
|
||||
b := New(16)
|
||||
byInst := b.Subscribe(Filter{InstanceID: "i1"})
|
||||
defer byInst.Close()
|
||||
byAgent := b.Subscribe(Filter{AgentID: "a2"})
|
||||
defer byAgent.Close()
|
||||
all := b.Subscribe(Filter{})
|
||||
defer all.Close()
|
||||
|
||||
b.Publish(logEvent("a1", "i1"))
|
||||
b.Publish(statsEvent("a2", "i2"))
|
||||
|
||||
if got := drain(byInst); got != 1 {
|
||||
t.Fatalf("InstanceID filter got %d, want 1", got)
|
||||
}
|
||||
if got := drain(byAgent); got != 1 {
|
||||
t.Fatalf("AgentID filter got %d, want 1", got)
|
||||
}
|
||||
if got := drain(all); got != 2 {
|
||||
t.Fatalf("unfiltered got %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDropOnFullBuffer(t *testing.T) {
|
||||
b := New(2)
|
||||
sub := b.Subscribe(Filter{})
|
||||
defer sub.Close()
|
||||
for i := 0; i < 5; i++ {
|
||||
b.Publish(statsEvent("a1", "i1"))
|
||||
}
|
||||
if got := drain(sub); got != 2 {
|
||||
t.Fatalf("buffered %d, want 2", got)
|
||||
}
|
||||
if d := sub.Dropped(); d != 3 {
|
||||
t.Fatalf("dropped %d, want 3", d)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
panelv1 "github.com/dbledeez/panel/proto/panel/v1"
|
||||
)
|
||||
|
||||
// LogBuffer is a bounded per-instance ring buffer of recent events. Its
|
||||
// purpose is UX: when the operator opens the Console tab for an instance,
|
||||
// we want them to see what's been going on — not just events that happen
|
||||
// *after* they subscribed. The buffer tails the event bus continuously so
|
||||
// a short history is always available.
|
||||
//
|
||||
// Events are stored verbatim (full *Event payload) so the caller can render
|
||||
// them exactly like the live SSE stream would. Only log, player, and
|
||||
// instance_state payloads are retained — stats and app_state churn too
|
||||
// fast and aren't what "Console history" should show.
|
||||
type LogBuffer struct {
|
||||
bus *Bus
|
||||
cap int
|
||||
mu sync.Mutex
|
||||
byInst map[string]*ringSlice
|
||||
stopped bool
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// NewLogBuffer creates the buffer and starts a background goroutine that
|
||||
// consumes the bus. Call Stop to shut down.
|
||||
//
|
||||
// capPerInstance is the maximum number of events kept per instance id. The
|
||||
// UI typically asks for the last 400, so 500 is a comfortable default.
|
||||
func NewLogBuffer(bus *Bus, capPerInstance int) *LogBuffer {
|
||||
if capPerInstance <= 0 {
|
||||
capPerInstance = 500
|
||||
}
|
||||
b := &LogBuffer{
|
||||
bus: bus,
|
||||
cap: capPerInstance,
|
||||
byInst: map[string]*ringSlice{},
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
go b.run()
|
||||
return b
|
||||
}
|
||||
|
||||
func (b *LogBuffer) run() {
|
||||
defer close(b.done)
|
||||
sub := b.bus.Subscribe(Filter{})
|
||||
defer sub.Close()
|
||||
for ev := range sub.Events() {
|
||||
if !isRetained(ev) {
|
||||
continue
|
||||
}
|
||||
id := eventInstanceID(ev)
|
||||
if id == "" {
|
||||
continue
|
||||
}
|
||||
b.mu.Lock()
|
||||
if b.stopped {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
r := b.byInst[id]
|
||||
if r == nil {
|
||||
r = newRing(b.cap)
|
||||
b.byInst[id] = r
|
||||
}
|
||||
r.push(ev)
|
||||
b.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// Recent returns up to `limit` most-recent retained events for an instance,
|
||||
// oldest first. Safe to call concurrently with Run.
|
||||
func (b *LogBuffer) Recent(instanceID string, limit int) []*panelv1.Event {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
r := b.byInst[instanceID]
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
return r.tail(limit)
|
||||
}
|
||||
|
||||
// Forget drops the buffer for an instance. Called when an instance is
|
||||
// deleted so memory doesn't leak for dead ids.
|
||||
func (b *LogBuffer) Forget(instanceID string) {
|
||||
b.mu.Lock()
|
||||
delete(b.byInst, instanceID)
|
||||
b.mu.Unlock()
|
||||
}
|
||||
|
||||
// Stop ends the background goroutine. Safe to call multiple times.
|
||||
func (b *LogBuffer) Stop() {
|
||||
b.mu.Lock()
|
||||
if b.stopped {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.stopped = true
|
||||
b.mu.Unlock()
|
||||
}
|
||||
|
||||
// isRetained decides whether an event belongs in the console history.
|
||||
//
|
||||
// - log lines: yes, that's the whole point.
|
||||
// - player events: yes — joins/leaves/chat/deaths belong on the console.
|
||||
// - instance_state: yes — "server started"-style transitions are context.
|
||||
// - stats / app_state: no — churn too fast; would push real content out
|
||||
// of the ring.
|
||||
func isRetained(e *panelv1.Event) bool {
|
||||
switch e.Payload.(type) {
|
||||
case *panelv1.Event_Log, *panelv1.Event_Player, *panelv1.Event_InstanceState:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ringSlice is a fixed-capacity FIFO. Implemented as a slice with head/len
|
||||
// tracking to avoid allocating on steady-state appends.
|
||||
type ringSlice struct {
|
||||
items []*panelv1.Event
|
||||
head int // index of the oldest entry (when len == cap)
|
||||
len int
|
||||
cap int
|
||||
}
|
||||
|
||||
func newRing(cap int) *ringSlice {
|
||||
return &ringSlice{items: make([]*panelv1.Event, cap), cap: cap}
|
||||
}
|
||||
|
||||
func (r *ringSlice) push(e *panelv1.Event) {
|
||||
if r.len < r.cap {
|
||||
r.items[(r.head+r.len)%r.cap] = e
|
||||
r.len++
|
||||
return
|
||||
}
|
||||
// At capacity: overwrite the oldest.
|
||||
r.items[r.head] = e
|
||||
r.head = (r.head + 1) % r.cap
|
||||
}
|
||||
|
||||
func (r *ringSlice) tail(limit int) []*panelv1.Event {
|
||||
if limit <= 0 || limit > r.len {
|
||||
limit = r.len
|
||||
}
|
||||
out := make([]*panelv1.Event, 0, limit)
|
||||
start := (r.head + r.len - limit + r.cap) % r.cap
|
||||
for i := 0; i < limit; i++ {
|
||||
out = append(out, r.items[(start+i)%r.cap])
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package events
|
||||
|
||||
// Replay ring: every published event gets a monotonic id and lands in a
|
||||
// fixed-capacity ring buffer so an SSE client that reconnects with
|
||||
// Last-Event-ID can be caught up without a full poll. The ring is
|
||||
// best-effort — when a client's resume point has already been overwritten,
|
||||
// Replay reports the gap and the caller falls back to the state snapshot
|
||||
// it sends on every connect anyway.
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
panelv1 "github.com/dbledeez/panel/proto/panel/v1"
|
||||
)
|
||||
|
||||
// Stamped pairs an event with its bus-assigned monotonic id. IDs start at 1
|
||||
// and never repeat for the lifetime of the process.
|
||||
type Stamped struct {
|
||||
ID uint64
|
||||
Event *panelv1.Event
|
||||
}
|
||||
|
||||
// DefaultRingCap is how many recent events the bus retains for replay.
|
||||
// State/app/player events are small and logs dominate volume; at a burst of
|
||||
// ~200 events/s this still covers ~20s of reconnect window, and the SSE
|
||||
// snapshot covers anything older.
|
||||
const DefaultRingCap = 4096
|
||||
|
||||
type replayRing struct {
|
||||
mu sync.Mutex
|
||||
buf []Stamped
|
||||
next uint64 // id to assign to the NEXT appended event (starts at 1)
|
||||
head int // index of oldest valid entry
|
||||
n int // number of valid entries (<= cap)
|
||||
}
|
||||
|
||||
func newReplayRing(capacity int) *replayRing {
|
||||
if capacity <= 0 {
|
||||
capacity = DefaultRingCap
|
||||
}
|
||||
return &replayRing{buf: make([]Stamped, capacity), next: 1}
|
||||
}
|
||||
|
||||
// append stamps e with the next monotonic id, stores it, and returns the id.
|
||||
func (r *replayRing) append(e *panelv1.Event) uint64 {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
id := r.next
|
||||
r.next++
|
||||
idx := (r.head + r.n) % len(r.buf)
|
||||
if r.n < len(r.buf) {
|
||||
r.buf[idx] = Stamped{ID: id, Event: e}
|
||||
r.n++
|
||||
} else {
|
||||
r.buf[r.head] = Stamped{ID: id, Event: e}
|
||||
r.head = (r.head + 1) % len(r.buf)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
// lastID returns the most recently assigned id (0 when nothing published).
|
||||
func (r *replayRing) lastID() uint64 {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.next - 1
|
||||
}
|
||||
|
||||
// since returns every retained event with ID > afterID, oldest first.
|
||||
// ok=false means the resume point has been overwritten (afterID is older
|
||||
// than the oldest retained event), so the returned slice — while still the
|
||||
// full retained suffix — has a gap before it; callers should resync from a
|
||||
// snapshot. afterID >= lastID returns (nil, true).
|
||||
func (r *replayRing) since(afterID uint64) (out []Stamped, ok bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.n == 0 {
|
||||
// Nothing retained: only gapless if the caller is already current.
|
||||
return nil, afterID >= r.next-1
|
||||
}
|
||||
oldest := r.buf[r.head].ID
|
||||
ok = afterID >= oldest-1
|
||||
for i := 0; i < r.n; i++ {
|
||||
s := r.buf[(r.head+i)%len(r.buf)]
|
||||
if s.ID > afterID {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out, ok
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
panelv1 "github.com/dbledeez/panel/proto/panel/v1"
|
||||
)
|
||||
|
||||
func stateEvent(instanceID string) *panelv1.Event {
|
||||
return &panelv1.Event{Payload: &panelv1.Event_InstanceState{
|
||||
InstanceState: &panelv1.InstanceStateUpdate{InstanceId: instanceID},
|
||||
}}
|
||||
}
|
||||
|
||||
func TestRingMonotonicIDsAndSince(t *testing.T) {
|
||||
r := newReplayRing(8)
|
||||
for i := 0; i < 5; i++ {
|
||||
id := r.append(stateEvent(fmt.Sprintf("i%d", i)))
|
||||
if want := uint64(i + 1); id != want {
|
||||
t.Fatalf("append id = %d, want %d", id, want)
|
||||
}
|
||||
}
|
||||
if r.lastID() != 5 {
|
||||
t.Fatalf("lastID = %d, want 5", r.lastID())
|
||||
}
|
||||
out, ok := r.since(2)
|
||||
if !ok {
|
||||
t.Fatalf("since(2) reported a gap on an unfilled ring")
|
||||
}
|
||||
if len(out) != 3 || out[0].ID != 3 || out[2].ID != 5 {
|
||||
t.Fatalf("since(2) = %+v, want ids 3..5", out)
|
||||
}
|
||||
// Fully caught up → empty, gapless.
|
||||
if out, ok := r.since(5); len(out) != 0 || !ok {
|
||||
t.Fatalf("since(lastID) = (%d events, ok=%v), want (0, true)", len(out), ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRingOverflowReportsGap(t *testing.T) {
|
||||
r := newReplayRing(4)
|
||||
for i := 0; i < 10; i++ { // ids 1..10; ring retains 7..10
|
||||
r.append(stateEvent("x"))
|
||||
}
|
||||
out, ok := r.since(3)
|
||||
if ok {
|
||||
t.Fatalf("since(3) after overflow should report a gap")
|
||||
}
|
||||
if len(out) != 4 || out[0].ID != 7 || out[3].ID != 10 {
|
||||
t.Fatalf("since(3) = %d events (first=%d), want retained suffix 7..10", len(out), out[0].ID)
|
||||
}
|
||||
// Resume point exactly at oldest-1 is gapless.
|
||||
if _, ok := r.since(6); !ok {
|
||||
t.Fatalf("since(oldest-1) should be gapless")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusReplayFiltersAndStampedSubscribe(t *testing.T) {
|
||||
b := New(16)
|
||||
sub := b.SubscribeStamped(Filter{InstanceID: "a"})
|
||||
defer sub.Close()
|
||||
|
||||
b.Publish(stateEvent("a")) // id 1
|
||||
b.Publish(stateEvent("b")) // id 2
|
||||
b.Publish(stateEvent("a")) // id 3
|
||||
|
||||
// Stamped subscription only sees matching events, with real ids.
|
||||
got1 := <-sub.Stamped()
|
||||
got2 := <-sub.Stamped()
|
||||
if got1.ID != 1 || got2.ID != 3 {
|
||||
t.Fatalf("stamped ids = %d,%d, want 1,3", got1.ID, got2.ID)
|
||||
}
|
||||
|
||||
// Replay applies the same filter.
|
||||
out, ok := b.Replay(0, Filter{InstanceID: "a"})
|
||||
if !ok || len(out) != 2 || out[0].ID != 1 || out[1].ID != 3 {
|
||||
t.Fatalf("Replay = %+v ok=%v, want ids 1,3 gapless", out, ok)
|
||||
}
|
||||
if b.LastID() != 3 {
|
||||
t.Fatalf("LastID = %d, want 3", b.LastID())
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusDroppedTotal(t *testing.T) {
|
||||
b := New(1)
|
||||
sub := b.Subscribe(Filter{})
|
||||
defer sub.Close()
|
||||
b.Publish(stateEvent("a")) // fills the 1-slot buffer
|
||||
b.Publish(stateEvent("a")) // dropped
|
||||
if b.DroppedTotal() != 1 {
|
||||
t.Fatalf("DroppedTotal = %d, want 1", b.DroppedTotal())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user