@@ -2,156 +2,182 @@ package app
22
33import (
44 "context"
5+ "fmt"
56 "sync"
67 "testing"
7− "testing/synctest"
88 "time"
99
1010 tea "charm.land/bubbletea/v2"
1111 "github.com/charmbracelet/crush/internal/pubsub"
1212 "github.com/stretchr/testify/require"
13− "go.uber.org/goleak"
1413)
1514
15+// TestSetupSubscriber_NormalFlow verifies that events published to the source
16+// broker are forwarded to the output broker.
1617func TestSetupSubscriber_NormalFlow(t *testing.T) {
17− synctest.Test(t, func(t *testing.T) {
18− f := newSubscriberFixture(t, 10)
18+ t.Parallel()
1919
20− time.Sleep(10 * time.Millisecond)
21− synctest.Wait()
20+ ctx, cancel := context.WithCancel(t.Context())
21+ defer cancel()
2222
23− f.broker.Publish(pubsub.CreatedEvent, "event1")
24− f.broker.Publish(pubsub.CreatedEvent, "event2")
23+ src := pubsub.NewBroker[string]()
24+ defer src.Shutdown()
25+ out := pubsub.NewBroker[tea.Msg]()
26+ defer out.Shutdown()
2527
26− for range 2 {
27− select {
28− case <-f.outputCh:
29− case <-time.After(5 * time.Second):
30− t.Fatal("Timed out waiting for messages")
31− }
28+ ch := out.Subscribe(ctx)
29+
30+ var wg sync.WaitGroup
31+ setupSubscriber(ctx, &wg, "test", src.Subscribe, out)
32+
33+ // Yield so the subscriber goroutine can call src.Subscribe before we publish.
34+ time.Sleep(10 * time.Millisecond)
35+
36+ src.Publish(pubsub.CreatedEvent, "hello")
37+ src.Publish(pubsub.CreatedEvent, "world")
38+
39+ for range 2 {
40+ select {
41+ case <-ch:
42+ case <-time.After(5 * time.Second):
43+ t.Fatal("timed out waiting for forwarded event")
3244 }
45+ }
3346
34− f.cancel()
35− f.wg.Wait()
36− })
47+ cancel()
48+ wg.Wait()
3749}
3850
39−func TestSetupSubscriber_SlowConsumer(t *testing.T) {
40− synctest.Test(t, func(t *testing.T) {
41− f := newSubscriberFixture(t, 0)
51+// TestSetupSubscriber_ContextCancellation verifies the goroutine exits cleanly
52+// when the context is cancelled.
53+func TestSetupSubscriber_ContextCancellation(t *testing.T) {
54+ t.Parallel()
4255
43− const numEvents = 5
56+ ctx, cancel := context.WithCancel(t.Context())
4457
45− var pubWg sync.WaitGroup
46− pubWg.Go(func() {
47− for range numEvents {
48− f.broker.Publish(pubsub.CreatedEvent, "event")
49− time.Sleep(10 * time.Millisecond)
50− synctest.Wait()
51− }
52− })
58+ src := pubsub.NewBroker[string]()
59+ defer src.Shutdown()
60+ out := pubsub.NewBroker[tea.Msg]()
61+ defer out.Shutdown()
5362
54− time.Sleep(time.Duration(numEvents) * (subscriberSendTimeout + 20*time.Millisecond))
55− synctest.Wait()
56−
57− received := 0
58− for {
59− select {
60− case <-f.outputCh:
61− received++
62− default:
63− pubWg.Wait()
64− f.cancel()
65− f.wg.Wait()
66− require.Less(t, received, numEvents, "Slow consumer should have dropped some messages")
67− return
68− }
69− }
70− })
63+ var wg sync.WaitGroup
64+ setupSubscriber(ctx, &wg, "test", src.Subscribe, out)
65+
66+ src.Publish(pubsub.CreatedEvent, "event")
67+ cancel()
68+
69+ done := make(chan struct{})
70+ go func() { wg.Wait(); close(done) }()
71+
72+ select {
73+ case <-done:
74+ case <-time.After(5 * time.Second):
75+ t.Fatal("setupSubscriber goroutine did not exit after context cancellation")
76+ }
7177}
7278
73−func TestSetupSubscriber_ContextCancellation(t *testing.T) {
74− synctest.Test(t, func(t *testing.T) {
75− f := newSubscriberFixture(t, 10)
79+// TestEvents_ZeroConsumers verifies that publishing with no subscribers does
80+// not block or panic.
81+func TestEvents_ZeroConsumers(t *testing.T) {
82+ t.Parallel()
83+
84+ broker := pubsub.NewBroker[tea.Msg]()
85+ defer broker.Shutdown()
86+
87+ require.Equal(t, 0, broker.GetSubscriberCount())
7688
77− f.broker.Publish(pubsub.CreatedEvent, "event1")
78− time.Sleep(100 * time.Millisecond)
79− synctest.Wait()
89+ // Must not block.
90+ done := make(chan struct{})
91+ go func() {
92+ broker.Publish(pubsub.UpdatedEvent, tea.Msg("msg1"))
93+ broker.Publish(pubsub.UpdatedEvent, tea.Msg("msg2"))
94+ close(done)
95+ }()
8096
81− f.cancel()
82− f.wg.Wait()
83− })
97+ select {
98+ case <-done:
99+ case <-time.After(time.Second):
100+ t.Fatal("Publish with zero consumers blocked")
101+ }
84102}
85103
86−func TestSetupSubscriber_DrainAfterDrop(t *testing.T) {
87− synctest.Test(t, func(t *testing.T) {
88− f := newSubscriberFixture(t, 0)
104+// TestEvents_OneConsumer verifies that a single subscriber receives every event
105+// exactly once.
106+func TestEvents_OneConsumer(t *testing.T) {
107+ t.Parallel()
89108
90− time.Sleep(10 * time.Millisecond)
91− synctest.Wait()
109+ ctx, cancel := context.WithCancel(t.Context())
110+ defer cancel()
92111
93− // First event: nobody reads outputCh so the timer fires (message dropped).
94− f.broker.Publish(pubsub.CreatedEvent, "event1")
95− time.Sleep(subscriberSendTimeout + 25*time.Millisecond)
96− synctest.Wait()
112+ broker := pubsub.NewBroker[tea.Msg]()
113+ defer broker.Shutdown()
97114
98− // Second event: triggers Stop()==false path; without the fix this deadlocks.
99− f.broker.Publish(pubsub.CreatedEvent, "event2")
115+ ch := broker.Subscribe(ctx)
100116
101− // If the timer drain deadlocks, wg.Wait never returns.
102− done := make(chan struct{})
103− go func() {
104− f.cancel()
105− f.wg.Wait()
106− close(done)
107− }()
117+ const n = 10
118+ for i := range n {
119+ broker.Publish(pubsub.UpdatedEvent, tea.Msg(i))
120+ }
108121
122+ for i := range n {
109123 select {
110− case <-done:
124+ case ev := <-ch:
125+ require.Equal(t, tea.Msg(i), ev.Payload)
111126 case <-time.After(5 * time.Second):
112− t.Fatal("setupSubscriber goroutine hung — likely timer drain deadlock")
127+ t.Fatalf("timed out waiting for event %d", i)
113128 }
114− })
129+ }
115130}
116131
117−func TestSetupSubscriber_NoTimerLeak(t *testing.T) {
118− defer goleak.VerifyNone(t)
119− synctest.Test(t, func(t *testing.T) {
120− f := newSubscriberFixture(t, 100)
132+// TestEvents_NConsumers verifies that every subscriber receives every event
133+// exactly once, regardless of how many concurrent consumers are attached.
134+func TestEvents_NConsumers(t *testing.T) {
135+ t.Parallel()
121136
122− for range 100 {
123− f.broker.Publish(pubsub.CreatedEvent, "event")
124− time.Sleep(5 * time.Millisecond)
125− synctest.Wait()
126− }
127−
128− f.cancel()
129− f.wg.Wait()
130− })
131−}
132−
133−type subscriberFixture struct {
134− broker *pubsub.Broker[string]
135− wg sync.WaitGroup
136− outputCh chan tea.Msg
137− cancel context.CancelFunc
137+ for _, n := range []int{2, 5, 10} {
138+ t.Run(fmt.Sprintf("consumers=%d", n), func(t *testing.T) {
139+ t.Parallel()
140+ testNConsumers(t, n)
141+ })
142+ }
138143}
139144
140−func newSubscriberFixture(t *testing.T, bufSize int) *subscriberFixture {
145+func testNConsumers(t *testing.T, n int) {
141146 t.Helper()
147+
142148 ctx, cancel := context.WithCancel(t.Context())
143− t.Cleanup(cancel)
149+ defer cancel()
150+
151+ broker := pubsub.NewBroker[tea.Msg]()
152+ defer broker.Shutdown()
144153
145− f := &subscriberFixture{
146− broker: pubsub.NewBroker[string](),
147− outputCh: make(chan tea.Msg, bufSize),
148− cancel: cancel,
154+ // Subscribe all N consumers before publishing.
155+ channels := make([]<-chan pubsub.Event[tea.Msg], n)
156+ for i := range n {
157+ channels[i] = broker.Subscribe(ctx)
149158 }
150− t.Cleanup(f.broker.Shutdown)
159+ require.Equal(t, n, broker.GetSubscriberCount())
151160
152− setupSubscriber(ctx, &f.wg, "test", func(ctx context.Context) <-chan pubsub.Event[string] {
153− return f.broker.Subscribe(ctx)
154− }, f.outputCh)
161+ const numEvents = 20
162+ for i := range numEvents {
163+ broker.Publish(pubsub.UpdatedEvent, tea.Msg(i))
164+ }
155165
156− return f
166+ // Each consumer must receive all numEvents messages.
167+ var wg sync.WaitGroup
168+ for i, ch := range channels {
169+ wg.Go(func() {
170+ for j := range numEvents {
171+ select {
172+ case ev := <-ch:
173+ require.Equal(t, tea.Msg(j), ev.Payload,
174+ "consumer %d: wrong payload for event %d", i, j)
175+ case <-time.After(5 * time.Second):
176+ t.Errorf("consumer %d: timed out waiting for event %d", i, j)
177+ return
178+ }
179+ }
180+ })
181+ }
182+ wg.Wait()
157183}