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
|
package recovery
import (
"strconv"
"sync"
"testing"
"time"
"github.com/centrifugal/protocol"
"github.com/stretchr/testify/require"
)
func TestPubSubSync(t *testing.T) {
psSync := NewPubSubSync()
require.Empty(t, psSync.subSync)
channels := []string{"ch1", "ch2", "ch3"}
var wg sync.WaitGroup
wg.Add(len(channels))
for _, channel := range channels {
go func(channel string) {
defer wg.Done()
done := make(chan struct{}, 1)
wait := make(chan struct{}, 1)
psSync.StartBuffering(channel)
go func() {
for i := 0; i < 10; i++ {
psSync.SyncPublication(channel, &protocol.Publication{}, func() {})
}
close(wait)
}()
go func() {
<-wait
psSync.LockBuffer(channel)
pubs := psSync.ReadBuffered(channel)
require.Equal(t, 10, len(pubs))
psSync.StopBuffering(channel)
close(done)
}()
select {
case <-done:
case <-time.After(time.Second):
require.Fail(t, "timeout")
}
}(channel)
}
wg.Wait()
require.Empty(t, psSync.subSync)
}
func BenchmarkPubSubSync(b *testing.B) {
psSync := NewPubSubSync()
var channels []string
for i := 0; i < 1; i++ {
channels = append(channels, "server-side-"+strconv.Itoa(i))
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
var wg sync.WaitGroup
wg.Add(len(channels))
for _, channel := range channels {
go func(channel string) {
defer wg.Done()
wait := make(chan struct{}, 1)
psSync.StartBuffering(channel)
go func() {
psSync.SyncPublication(channel, &protocol.Publication{}, func() {})
close(wait)
}()
<-wait
psSync.LockBuffer(channel)
_ = psSync.ReadBuffered(channel)
psSync.StopBuffering(channel)
}(channel)
}
wg.Wait()
}
b.StopTimer()
b.ReportAllocs()
}
|
