diff --git a/internal/master/events/broker.go b/internal/master/events/broker.go new file mode 100644 index 0000000..994fe51 --- /dev/null +++ b/internal/master/events/broker.go @@ -0,0 +1,56 @@ +// Package events fans out "something changed" signals from the agent hub to +// open browser streams. +package events + +import "sync" + +// Topic names what changed. Subscribers refetch that part of the UI. +type Topic string + +// Topics. +const ( + Services Topic = "services" + Agents Topic = "agents" +) + +// Broker is an in-process publish/subscribe hub. +type Broker struct { + mu sync.Mutex + subs map[chan Topic]struct{} +} + +// NewBroker returns an empty Broker. +func NewBroker() *Broker { + return &Broker{subs: make(map[chan Topic]struct{})} +} + +// Subscribe returns a channel of topics and a cancel function that must be +// called when the subscriber goes away. Cancel is safe to call twice. +func (b *Broker) Subscribe() (<-chan Topic, func()) { + ch := make(chan Topic, 16) + b.mu.Lock() + b.subs[ch] = struct{}{} + b.mu.Unlock() + var once sync.Once + return ch, func() { + once.Do(func() { + b.mu.Lock() + delete(b.subs, ch) + b.mu.Unlock() + }) + } +} + +// Publish signals t to every subscriber. A subscriber whose buffer is full is +// skipped rather than blocking the agent hub; it already has signals queued +// and will refetch current state when it processes them. +func (b *Broker) Publish(t Topic) { + b.mu.Lock() + defer b.mu.Unlock() + for ch := range b.subs { + select { + case ch <- t: + default: + } + } +} diff --git a/internal/master/events/broker_test.go b/internal/master/events/broker_test.go new file mode 100644 index 0000000..7613051 --- /dev/null +++ b/internal/master/events/broker_test.go @@ -0,0 +1,48 @@ +package events + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestBroker_DeliversToAllSubscribers(t *testing.T) { + b := NewBroker() + a, cancelA := b.Subscribe() + defer cancelA() + c, cancelC := b.Subscribe() + defer cancelC() + + b.Publish(Services) + assert.Equal(t, Services, <-a) + assert.Equal(t, Services, <-c) +} + +func TestBroker_CancelStopsDelivery(t *testing.T) { + b := NewBroker() + ch, cancel := b.Subscribe() + cancel() + cancel() // safe to call twice + b.Publish(Agents) + assert.Empty(t, ch) +} + +func TestBroker_SlowSubscriberDoesNotBlockPublish(t *testing.T) { + b := NewBroker() + _, cancel := b.Subscribe() + defer cancel() + + done := make(chan struct{}) + go func() { + defer close(done) + for range 1000 { + b.Publish(Services) // must return even though nobody reads + } + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Publish blocked on a slow subscriber") + } +}