-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathps_test.go
119 lines (92 loc) · 3.45 KB
/
ps_test.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
package ps_test
import (
"testing"
"github.com/peterbourgon/ps"
)
func TestBasics(t *testing.T) {
t.Parallel()
t.Run("no subscribers", func(t *testing.T) {
broker := ps.NewBroker[int]()
compareStats(t, broker.Publish(1), ps.Stats{})
compareStats(t, broker.Publish(2), ps.Stats{})
compareStats(t, broker.Publish(3), ps.Stats{})
})
t.Run("skip subscriber", func(t *testing.T) {
broker := ps.NewBroker[int]()
c := make(chan int)
broker.Subscribe(c, func(int) bool { return false })
compareStats(t, broker.Publish(1), ps.Stats{Skips: 1})
compareStats(t, broker.Publish(2), ps.Stats{Skips: 1})
compareStats(t, broker.Publish(3), ps.Stats{Skips: 1})
stats, err := broker.Unsubscribe(c)
requireNoError(t, err)
compareStats(t, stats, ps.Stats{Skips: 3})
})
t.Run("slow subscriber", func(t *testing.T) {
broker := ps.NewBroker[int]()
c1, c3 := make(chan int, 1), make(chan int, 3)
broker.Subscribe(c1, func(int) bool { return true })
broker.Subscribe(c3, func(int) bool { return true })
compareStats(t, broker.Publish(1), ps.Stats{Sends: 2})
compareStats(t, broker.Publish(2), ps.Stats{Sends: 1, Drops: 1})
compareStats(t, broker.Publish(3), ps.Stats{Sends: 1, Drops: 1})
compareStats(t, broker.Publish(4), ps.Stats{Sends: 0, Drops: 2})
compareStats(t, broker.Publish(5), ps.Stats{Sends: 0, Drops: 2})
expectEqual(t, 1, <-c1)
expectEqual(t, 1, <-c3)
compareStats(t, broker.Publish(6), ps.Stats{Sends: 2, Drops: 0})
compareStats(t, broker.Publish(7), ps.Stats{Sends: 0, Drops: 2})
c1s, err := broker.Unsubscribe(c1)
requireNoError(t, err)
compareStats(t, c1s, ps.Stats{Skips: 0, Sends: 2, Drops: 5})
c3s, err := broker.Unsubscribe(c3)
requireNoError(t, err)
compareStats(t, c3s, ps.Stats{Skips: 0, Sends: 4, Drops: 3})
})
t.Run("subscriber stats", func(t *testing.T) {
broker := ps.NewBroker[int]()
mod2, mod3 := make(chan int, 100), make(chan int, 100)
broker.Subscribe(mod2, func(i int) bool { return i%2 == 0 })
broker.Subscribe(mod3, func(i int) bool { return i%3 == 0 })
compareStats(t, broker.Publish(1), ps.Stats{Skips: 2, Sends: 0})
compareStats(t, broker.Publish(2), ps.Stats{Skips: 1, Sends: 1})
compareStats(t, broker.Publish(3), ps.Stats{Skips: 1, Sends: 1})
compareStats(t, broker.Publish(4), ps.Stats{Skips: 1, Sends: 1})
compareStats(t, broker.Publish(5), ps.Stats{Skips: 2, Sends: 0})
compareStats(t, broker.Publish(6), ps.Stats{Skips: 0, Sends: 2})
mod2stats, err := broker.Stats(mod2)
requireNoError(t, err)
compareStats(t, mod2stats, ps.Stats{Skips: 3, Sends: 3})
mod3stats, err := broker.Stats(mod3)
requireNoError(t, err)
compareStats(t, mod3stats, ps.Stats{Skips: 4, Sends: 2})
allstats := broker.ActiveSubscribers()
expectEqual(t, 2, len(allstats))
compareStats(t, allstats[0], mod2stats)
compareStats(t, allstats[1], mod3stats)
mod2unsub, err := broker.Unsubscribe(mod2)
requireNoError(t, err)
compareStats(t, mod2stats, mod2unsub)
mod3unsub, err := broker.Unsubscribe(mod3)
requireNoError(t, err)
compareStats(t, mod3stats, mod3unsub)
})
}
func compareStats(tb testing.TB, have, want ps.Stats) {
tb.Helper()
if have.String() != want.String() {
tb.Errorf("stats: have %+v, want %+v", have, want)
}
}
func requireNoError(tb testing.TB, err error) {
tb.Helper()
if err != nil {
tb.Fatalf("fatal error: %v", err)
}
}
func expectEqual[T comparable](tb testing.TB, want, have T) {
tb.Helper()
if want != have {
tb.Errorf("want %+v, have %+v", want, have)
}
}