1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package eventbus_test
4
5import (
6 "context"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/assert"
12 "github.com/stretchr/testify/require"
13
14 "bnema.dev/ermite/internal/adapters/eventbus"
15 "bnema.dev/ermite/internal/domain"
16)
17
18func ev(ref string) domain.Event {
19 return domain.PushEvent{Updates: []domain.RefUpdate{{Ref: ref}}}
20}
21
22func recv(t *testing.T, ch <-chan domain.Event) domain.Event {
23 t.Helper()
24 select {
25 case e, ok := <-ch:
26 require.True(t, ok, "channel closed")
27 return e
28 case <-time.After(2 * time.Second):
29 t.Fatal("no event")
30 return nil
31 }
32}
33
34func TestFanOut(t *testing.T) {
35 ctx, cancel := context.WithCancel(context.Background())
36 defer cancel()
37 b := eventbus.New()
38 a, c := b.Subscribe(ctx), b.Subscribe(ctx)
39
40 b.Publish(ctx, ev("1"))
41 b.Publish(ctx, domain.RepoCreated{})
42 assert.Equal(t, ev("1"), recv(t, a))
43 assert.Equal(t, ev("1"), recv(t, c))
44 assert.Equal(t, "repo.created", recv(t, a).EventKind())
45 assert.Equal(t, "repo.created", recv(t, c).EventKind())
46}
47
48func TestOnlyEventsFromSubscription(t *testing.T) {
49 ctx, cancel := context.WithCancel(context.Background())
50 defer cancel()
51 b := eventbus.New()
52 b.Publish(ctx, ev("before"))
53 ch := b.Subscribe(ctx)
54 b.Publish(ctx, ev("after"))
55 assert.Equal(t, ev("after"), recv(t, ch))
56}
57
58func TestSlowSubscriberGetsGapAndDoesNotBlock(t *testing.T) {
59 ctx, cancel := context.WithCancel(context.Background())
60 defer cancel()
61 b := eventbus.NewWithBuffer(4)
62 slow, fast := b.Subscribe(ctx), b.Subscribe(ctx)
63
64 done := make(chan struct{})
65 go func() {
66 defer close(done)
67 for i := range 20 {
68 b.Publish(ctx, ev(string(rune('a'+i))))
69 // The fast subscriber keeps up.
70 recv(t, fast)
71 }
72 }()
73 select {
74 case <-done:
75 case <-time.After(5 * time.Second):
76 t.Fatal("Publish blocked on a slow subscriber")
77 }
78
79 // The slow one lost events: it sees a gap marker, then only recent events.
80 var got []domain.Event
81 for len(slow) > 0 {
82 got = append(got, <-slow)
83 }
84 require.NotEmpty(t, got)
85 var gapAt = -1
86 for i, e := range got {
87 if _, ok := e.(domain.EventGap); ok {
88 gapAt = i
89 }
90 }
91 assert.GreaterOrEqual(t, gapAt, 0, "a gap marker is delivered")
92 assert.Equal(t, ev("t"), got[len(got)-1], "the latest event is delivered")
93 assert.LessOrEqual(t, len(got), 4)
94}
95
96func TestUnsubscribeOnContext(t *testing.T) {
97 b := eventbus.New()
98 ctx, cancel := context.WithCancel(context.Background())
99 ch := b.Subscribe(ctx)
100 require.Equal(t, 1, b.Subscribers())
101 cancel()
102 select {
103 case _, ok := <-ch:
104 assert.False(t, ok)
105 case <-time.After(2 * time.Second):
106 t.Fatal("channel not closed")
107 }
108 assert.Equal(t, 0, b.Subscribers())
109 b.Publish(context.Background(), ev("x")) // no panic
110}
111
112func TestConcurrentPublishSubscribe(t *testing.T) {
113 b := eventbus.NewWithBuffer(8)
114 ctx, cancel := context.WithCancel(context.Background())
115 var wg sync.WaitGroup
116 for range 4 {
117 wg.Add(2)
118 go func() {
119 defer wg.Done()
120 for range 200 {
121 b.Publish(ctx, ev("x"))
122 }
123 }()
124 go func() {
125 defer wg.Done()
126 sctx, scancel := context.WithCancel(ctx)
127 ch := b.Subscribe(sctx)
128 for range 5 {
129 select {
130 case <-ch:
131 case <-time.After(10 * time.Millisecond):
132 }
133 }
134 scancel()
135 }()
136 }
137 wg.Wait()
138 cancel()
139}