feat(master): in-process broker for live UI change signals

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:
+		}
+	}
+}

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")
+	}
+}