diff --git a/internal/web/broadcaster.go b/internal/web/broadcaster.go new file mode 100644 index 0000000..4a61a7b --- /dev/null +++ b/internal/web/broadcaster.go @@ -0,0 +1,42 @@ +package web + +import "sync" + +// Broadcaster fans a string message out to all SSE subscribers. Slow +// subscribers drop messages rather than blocking publishers. +type Broadcaster struct { + mu sync.Mutex + subs map[chan string]struct{} +} + +func NewBroadcaster() *Broadcaster { + return &Broadcaster{subs: make(map[chan string]struct{})} +} + +func (b *Broadcaster) Subscribe() chan string { + ch := make(chan string, 8) + b.mu.Lock() + b.subs[ch] = struct{}{} + b.mu.Unlock() + return ch +} + +func (b *Broadcaster) Unsubscribe(ch chan string) { + b.mu.Lock() + if _, ok := b.subs[ch]; ok { + delete(b.subs, ch) + close(ch) + } + b.mu.Unlock() +} + +func (b *Broadcaster) Publish(msg string) { + b.mu.Lock() + defer b.mu.Unlock() + for ch := range b.subs { + select { + case ch <- msg: + default: // drop for slow subscriber + } + } +} diff --git a/internal/web/broadcaster_test.go b/internal/web/broadcaster_test.go new file mode 100644 index 0000000..f83227c --- /dev/null +++ b/internal/web/broadcaster_test.go @@ -0,0 +1,42 @@ +package web + +import ( + "testing" + "time" +) + +func TestBroadcasterDeliversToSubscribers(t *testing.T) { + b := NewBroadcaster() + ch := b.Subscribe() + defer b.Unsubscribe(ch) + + b.Publish("hello") + + select { + case msg := <-ch: + if msg != "hello" { + t.Fatalf("got %q want hello", msg) + } + case <-time.After(time.Second): + t.Fatal("subscriber did not receive message") + } +} + +func TestBroadcasterDropsWhenSubscriberFull(t *testing.T) { + b := NewBroadcaster() + ch := b.Subscribe() + defer b.Unsubscribe(ch) + // Publishing many messages without reading must not block (slow client). + done := make(chan struct{}) + go func() { + for i := 0; i < 100; i++ { + b.Publish("x") + } + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Publish blocked on a full subscriber") + } +}