1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package eventbus_test
4
5import (
6 "context"
7 "runtime"
8 "sync"
9 "testing"
10 "testing/synctest"
11
12 "github.com/stretchr/testify/assert"
13 "github.com/stretchr/testify/require"
14
15 "bnema.dev/ermite/internal/adapters/eventbus"
16 "bnema.dev/ermite/internal/domain"
17)
18
19func ev(ref string) domain.Event {
20 return domain.PushEvent{Updates: []domain.RefUpdate{{Ref: ref}}}
21}
22
23// recv takes the event a subscriber already has. Publish delivers into the
24// buffer before it returns, so there is nothing to wait for.
25func recv(t *testing.T, ch <-chan domain.Event) domain.Event {
26 t.Helper()
27 synctest.Wait()
28 select {
29 case e, ok := <-ch:
30 require.True(t, ok, "channel closed")
31 return e
32 default:
33 t.Fatal("no event")
34 return nil
35 }
36}
37
38func TestFanOut(t *testing.T) {
39 t.Parallel()
40 synctest.Test(t, func(t *testing.T) {
41 ctx, cancel := context.WithCancel(context.Background())
42 defer cancel()
43 b := eventbus.New()
44 a, c := b.Subscribe(ctx), b.Subscribe(ctx)
45
46 b.Publish(ctx, ev("1"))
47 b.Publish(ctx, domain.RepoCreated{})
48 assert.Equal(t, ev("1"), recv(t, a))
49 assert.Equal(t, ev("1"), recv(t, c))
50 assert.Equal(t, "repo.created", recv(t, a).EventKind())
51 assert.Equal(t, "repo.created", recv(t, c).EventKind())
52 })
53}
54
55func TestOnlyEventsFromSubscription(t *testing.T) {
56 t.Parallel()
57 synctest.Test(t, func(t *testing.T) {
58 ctx, cancel := context.WithCancel(context.Background())
59 defer cancel()
60 b := eventbus.New()
61 b.Publish(ctx, ev("before"))
62 ch := b.Subscribe(ctx)
63 b.Publish(ctx, ev("after"))
64 assert.Equal(t, ev("after"), recv(t, ch))
65 synctest.Wait()
66 assert.Empty(t, ch, "nothing else is buffered")
67 })
68}
69
70func TestSlowSubscriberGetsGapAndDoesNotBlock(t *testing.T) {
71 t.Parallel()
72 synctest.Test(t, func(t *testing.T) {
73 ctx, cancel := context.WithCancel(context.Background())
74 defer cancel()
75 b := eventbus.NewWithBuffer(4)
76 slow, fast := b.Subscribe(ctx), b.Subscribe(ctx)
77
78 // Publish runs in this goroutine: one that blocked on the slow
79 // subscriber would deadlock the bubble and fail the test.
80 for i := range 20 {
81 b.Publish(ctx, ev(string(rune('a'+i))))
82 // The fast subscriber keeps up.
83 assert.Equal(t, ev(string(rune('a'+i))), recv(t, fast))
84 }
85
86 // The slow one lost events: it sees a gap marker, then only recent events.
87 var got []domain.Event
88 for len(slow) > 0 {
89 got = append(got, <-slow)
90 }
91 require.NotEmpty(t, got)
92 var gapAt = -1
93 for i, e := range got {
94 if _, ok := e.(domain.EventGap); ok {
95 gapAt = i
96 }
97 }
98 assert.GreaterOrEqual(t, gapAt, 0, "a gap marker is delivered")
99 assert.Equal(t, ev("t"), got[len(got)-1], "the latest event is delivered")
100 assert.LessOrEqual(t, len(got), 4)
101 })
102}
103
104func TestUnsubscribeOnContext(t *testing.T) {
105 t.Parallel()
106 synctest.Test(t, func(t *testing.T) {
107 b := eventbus.New()
108 ctx, cancel := context.WithCancel(context.Background())
109 ch := b.Subscribe(ctx)
110 require.Equal(t, 1, b.Subscribers())
111 synctest.Wait()
112 require.Equal(t, 1, b.Subscribers(), "still subscribed while ctx lives")
113 cancel()
114 synctest.Wait() // the AfterFunc of the context has run
115 select {
116 case _, ok := <-ch:
117 assert.False(t, ok)
118 default:
119 t.Fatal("channel not closed")
120 }
121 assert.Equal(t, 0, b.Subscribers())
122 b.Publish(context.Background(), ev("x")) // no panic
123 })
124}
125
126func TestConcurrentPublishSubscribe(t *testing.T) {
127 t.Parallel()
128 synctest.Test(t, func(t *testing.T) {
129 b := eventbus.NewWithBuffer(8)
130 ctx, cancel := context.WithCancel(context.Background())
131 var wg sync.WaitGroup
132 for range 4 {
133 wg.Add(2)
134 go func() {
135 defer wg.Done()
136 for range 200 {
137 b.Publish(ctx, ev("x"))
138 }
139 }()
140 go func() {
141 defer wg.Done()
142 sctx, scancel := context.WithCancel(ctx)
143 ch := b.Subscribe(sctx)
144 for range 5 {
145 select {
146 case <-ch:
147 default: // nothing buffered yet: Publish never blocks, nor does this
148 }
149 runtime.Gosched()
150 }
151 scancel()
152 }()
153 }
154 wg.Wait()
155 cancel()
156 synctest.Wait()
157 assert.Zero(t, b.Subscribers(), "every subscription ended")
158 })
159}