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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
package redis
import (
"sync"
"testing"
"time"
"github.com/rafaeljusto/redigomock"
"github.com/stretchr/testify/require"
)
const (
runnerKey = "runner:build_queue:10"
)
func createSubscriptionMessage(key, data string) []interface{} {
return []interface{}{
[]byte("message"),
[]byte(key),
[]byte(data),
}
}
func createSubscribeMessage(key string) []interface{} {
return []interface{}{
[]byte("subscribe"),
[]byte(key),
[]byte("1"),
}
}
func createUnsubscribeMessage(key string) []interface{} {
return []interface{}{
[]byte("unsubscribe"),
[]byte(key),
[]byte("1"),
}
}
func countWatchers(key string) int {
keyWatcherMutex.Lock()
defer keyWatcherMutex.Unlock()
return len(keyWatcher[key])
}
func deleteWatchers(key string) {
keyWatcherMutex.Lock()
defer keyWatcherMutex.Unlock()
delete(keyWatcher, key)
}
// Forces a run of the `Process` loop against a mock PubSubConn.
func processMessages(numWatchers int, value string) {
psc := redigomock.NewConn()
// Setup the initial subscription message
psc.Command("SUBSCRIBE", keySubChannel).Expect(createSubscribeMessage(keySubChannel))
psc.Command("UNSUBSCRIBE", keySubChannel).Expect(createUnsubscribeMessage(keySubChannel))
psc.AddSubscriptionMessage(createSubscriptionMessage(keySubChannel, runnerKey+"="+value))
// Wait for all the `WatchKey` calls to be registered
for countWatchers(runnerKey) != numWatchers {
time.Sleep(time.Millisecond)
}
processInner(psc)
}
func TestWatchKeySeenChange(t *testing.T) {
conn, td := setupMockPool()
defer td()
conn.Command("GET", runnerKey).Expect("something")
wg := &sync.WaitGroup{}
wg.Add(1)
go func() {
val, err := WatchKey(runnerKey, "something", time.Second)
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusSeenChange, val, "Expected value to change")
wg.Done()
}()
processMessages(1, "somethingelse")
wg.Wait()
}
func TestWatchKeyNoChange(t *testing.T) {
conn, td := setupMockPool()
defer td()
conn.Command("GET", runnerKey).Expect("something")
wg := &sync.WaitGroup{}
wg.Add(1)
go func() {
val, err := WatchKey(runnerKey, "something", time.Second)
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusNoChange, val, "Expected notification without change to value")
wg.Done()
}()
processMessages(1, "something")
wg.Wait()
}
func TestWatchKeyTimeout(t *testing.T) {
conn, td := setupMockPool()
defer td()
conn.Command("GET", runnerKey).Expect("something")
val, err := WatchKey(runnerKey, "something", time.Millisecond)
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusTimeout, val, "Expected value to not change")
// Clean up watchers since Process isn't doing that for us (not running)
deleteWatchers(runnerKey)
}
func TestWatchKeyAlreadyChanged(t *testing.T) {
conn, td := setupMockPool()
defer td()
conn.Command("GET", runnerKey).Expect("somethingelse")
val, err := WatchKey(runnerKey, "something", time.Second)
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusAlreadyChanged, val, "Expected value to have already changed")
// Clean up watchers since Process isn't doing that for us (not running)
deleteWatchers(runnerKey)
}
func TestWatchKeyMassivelyParallel(t *testing.T) {
runTimes := 100 // 100 parallel watchers
conn, td := setupMockPool()
defer td()
wg := &sync.WaitGroup{}
wg.Add(runTimes)
getCmd := conn.Command("GET", runnerKey)
for i := 0; i < runTimes; i++ {
getCmd = getCmd.Expect("something")
}
for i := 0; i < runTimes; i++ {
go func() {
val, err := WatchKey(runnerKey, "something", time.Second)
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusSeenChange, val, "Expected value to change")
wg.Done()
}()
}
processMessages(runTimes, "somethingelse")
wg.Wait()
}
|
