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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
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()
}
func TestShutdown(t *testing.T) {
conn, td := setupMockPool()
defer td()
defer func() { shutdown = make(chan struct{}) }()
conn.Command("GET", runnerKey).Expect("something")
wg := &sync.WaitGroup{}
wg.Add(2)
go func() {
val, err := WatchKey(runnerKey, "something", 10*time.Second)
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusNoChange, val, "Expected value not to change")
wg.Done()
}()
go func() {
require.Eventually(t, func() bool { return countWatchers(runnerKey) == 1 }, 10*time.Second, time.Millisecond)
Shutdown()
wg.Done()
}()
wg.Wait()
require.Eventually(t, func() bool { return countWatchers(runnerKey) == 0 }, 10*time.Second, time.Millisecond)
// Adding a key after the shutdown should result in an immediate response
var val WatchKeyStatus
var err error
done := make(chan struct{})
go func() {
val, err = WatchKey(runnerKey, "something", 10*time.Second)
close(done)
}()
select {
case <-done:
require.NoError(t, err, "Expected no error")
require.Equal(t, WatchKeyStatusNoChange, val, "Expected value not to change")
case <-time.After(100 * time.Millisecond):
t.Fatal("timeout waiting for WatchKey")
}
}
|
