1 package redis_test
2
3 import (
4 "bytes"
5 "net"
6 "time"
7
8 "github.com/go-redis/redis"
9
10 . "github.com/onsi/ginkgo"
11 . "github.com/onsi/gomega"
12 )
13
14 var _ = Describe("Client", func() {
15 var client *redis.Client
16
17 BeforeEach(func() {
18 client = redis.NewClient(redisOptions())
19 Expect(client.FlushDB().Err()).NotTo(HaveOccurred())
20 })
21
22 AfterEach(func() {
23 client.Close()
24 })
25
26 It("should Stringer", func() {
27 Expect(client.String()).To(Equal("Redis<:6380 db:15>"))
28 })
29
30 It("should ping", func() {
31 val, err := client.Ping().Result()
32 Expect(err).NotTo(HaveOccurred())
33 Expect(val).To(Equal("PONG"))
34 })
35
36 It("should return pool stats", func() {
37 Expect(client.PoolStats()).To(BeAssignableToTypeOf(&redis.PoolStats{}))
38 })
39
40 It("should support custom dialers", func() {
41 custom := redis.NewClient(&redis.Options{
42 Addr: ":1234",
43 Dialer: func() (net.Conn, error) {
44 return net.Dial("tcp", redisAddr)
45 },
46 })
47
48 val, err := custom.Ping().Result()
49 Expect(err).NotTo(HaveOccurred())
50 Expect(val).To(Equal("PONG"))
51 Expect(custom.Close()).NotTo(HaveOccurred())
52 })
53
54 It("should close", func() {
55 Expect(client.Close()).NotTo(HaveOccurred())
56 err := client.Ping().Err()
57 Expect(err).To(MatchError("redis: client is closed"))
58 })
59
60 It("should close pubsub without closing the client", func() {
61 pubsub := client.Subscribe()
62 Expect(pubsub.Close()).NotTo(HaveOccurred())
63
64 _, err := pubsub.Receive()
65 Expect(err).To(MatchError("redis: client is closed"))
66 Expect(client.Ping().Err()).NotTo(HaveOccurred())
67 })
68
69 It("should close Tx without closing the client", func() {
70 err := client.Watch(func(tx *redis.Tx) error {
71 _, err := tx.Pipelined(func(pipe redis.Pipeliner) error {
72 pipe.Ping()
73 return nil
74 })
75 return err
76 })
77 Expect(err).NotTo(HaveOccurred())
78
79 Expect(client.Ping().Err()).NotTo(HaveOccurred())
80 })
81
82 It("should close pipeline without closing the client", func() {
83 pipeline := client.Pipeline()
84 Expect(pipeline.Close()).NotTo(HaveOccurred())
85
86 pipeline.Ping()
87 _, err := pipeline.Exec()
88 Expect(err).To(MatchError("redis: client is closed"))
89
90 Expect(client.Ping().Err()).NotTo(HaveOccurred())
91 })
92
93 It("should close pubsub when client is closed", func() {
94 pubsub := client.Subscribe()
95 Expect(client.Close()).NotTo(HaveOccurred())
96
97 _, err := pubsub.Receive()
98 Expect(err).To(MatchError("redis: client is closed"))
99
100 Expect(pubsub.Close()).NotTo(HaveOccurred())
101 })
102
103 It("should close pipeline when client is closed", func() {
104 pipeline := client.Pipeline()
105 Expect(client.Close()).NotTo(HaveOccurred())
106 Expect(pipeline.Close()).NotTo(HaveOccurred())
107 })
108
109 It("should select DB", func() {
110 db2 := redis.NewClient(&redis.Options{
111 Addr: redisAddr,
112 DB: 2,
113 })
114 Expect(db2.FlushDB().Err()).NotTo(HaveOccurred())
115 Expect(db2.Get("db").Err()).To(Equal(redis.Nil))
116 Expect(db2.Set("db", 2, 0).Err()).NotTo(HaveOccurred())
117
118 n, err := db2.Get("db").Int64()
119 Expect(err).NotTo(HaveOccurred())
120 Expect(n).To(Equal(int64(2)))
121
122 Expect(client.Get("db").Err()).To(Equal(redis.Nil))
123
124 Expect(db2.FlushDB().Err()).NotTo(HaveOccurred())
125 Expect(db2.Close()).NotTo(HaveOccurred())
126 })
127
128 It("processes custom commands", func() {
129 cmd := redis.NewCmd("PING")
130 client.Process(cmd)
131
132
133 Expect(client.Echo("hello").Err()).NotTo(HaveOccurred())
134
135 Expect(cmd.Err()).NotTo(HaveOccurred())
136 Expect(cmd.Val()).To(Equal("PONG"))
137 })
138
139 It("should retry command on network error", func() {
140 Expect(client.Close()).NotTo(HaveOccurred())
141
142 client = redis.NewClient(&redis.Options{
143 Addr: redisAddr,
144 MaxRetries: 1,
145 })
146
147
148 cn, err := client.Pool().Get()
149 Expect(err).NotTo(HaveOccurred())
150
151 cn.SetNetConn(&badConn{})
152 client.Pool().Put(cn)
153
154 err = client.Ping().Err()
155 Expect(err).NotTo(HaveOccurred())
156 })
157
158 It("should retry with backoff", func() {
159 clientNoRetry := redis.NewClient(&redis.Options{
160 Addr: ":1234",
161 MaxRetries: 0,
162 })
163 defer clientNoRetry.Close()
164
165 clientRetry := redis.NewClient(&redis.Options{
166 Addr: ":1234",
167 MaxRetries: 5,
168 MaxRetryBackoff: 128 * time.Millisecond,
169 })
170 defer clientRetry.Close()
171
172 startNoRetry := time.Now()
173 err := clientNoRetry.Ping().Err()
174 Expect(err).To(HaveOccurred())
175 elapseNoRetry := time.Since(startNoRetry)
176
177 startRetry := time.Now()
178 err = clientRetry.Ping().Err()
179 Expect(err).To(HaveOccurred())
180 elapseRetry := time.Since(startRetry)
181
182 Expect(elapseRetry).To(BeNumerically(">", elapseNoRetry, 10*time.Millisecond))
183 })
184
185 It("should update conn.UsedAt on read/write", func() {
186 cn, err := client.Pool().Get()
187 Expect(err).NotTo(HaveOccurred())
188 Expect(cn.UsedAt).NotTo(BeZero())
189 createdAt := cn.UsedAt()
190
191 client.Pool().Put(cn)
192 Expect(cn.UsedAt().Equal(createdAt)).To(BeTrue())
193
194 err = client.Ping().Err()
195 Expect(err).NotTo(HaveOccurred())
196
197 cn, err = client.Pool().Get()
198 Expect(err).NotTo(HaveOccurred())
199 Expect(cn).NotTo(BeNil())
200 Expect(cn.UsedAt().After(createdAt)).To(BeTrue())
201 })
202
203 It("should process command with special chars", func() {
204 set := client.Set("key", "hello1\r\nhello2\r\n", 0)
205 Expect(set.Err()).NotTo(HaveOccurred())
206 Expect(set.Val()).To(Equal("OK"))
207
208 get := client.Get("key")
209 Expect(get.Err()).NotTo(HaveOccurred())
210 Expect(get.Val()).To(Equal("hello1\r\nhello2\r\n"))
211 })
212
213 It("should handle big vals", func() {
214 bigVal := bytes.Repeat([]byte{'*'}, 2e6)
215
216 err := client.Set("key", bigVal, 0).Err()
217 Expect(err).NotTo(HaveOccurred())
218
219
220 Expect(client.Close()).NotTo(HaveOccurred())
221 client = redis.NewClient(redisOptions())
222
223 got, err := client.Get("key").Bytes()
224 Expect(err).NotTo(HaveOccurred())
225 Expect(got).To(Equal(bigVal))
226 })
227
228 It("should call WrapProcess", func() {
229 var fnCalled bool
230
231 client.WrapProcess(func(old func(redis.Cmder) error) func(redis.Cmder) error {
232 return func(cmd redis.Cmder) error {
233 fnCalled = true
234 return old(cmd)
235 }
236 })
237
238 Expect(client.Ping().Err()).NotTo(HaveOccurred())
239 Expect(fnCalled).To(BeTrue())
240 })
241
242 It("should call WrapProcess after WithContext", func() {
243 var fn1Called, fn2Called bool
244
245 client.WrapProcess(func(old func(cmd redis.Cmder) error) func(cmd redis.Cmder) error {
246 return func(cmd redis.Cmder) error {
247 fn1Called = true
248 return old(cmd)
249 }
250 })
251
252 client2 := client.WithContext(client.Context())
253 client2.WrapProcess(func(old func(cmd redis.Cmder) error) func(cmd redis.Cmder) error {
254 return func(cmd redis.Cmder) error {
255 fn2Called = true
256 return old(cmd)
257 }
258 })
259
260 Expect(client2.Ping().Err()).NotTo(HaveOccurred())
261 Expect(fn2Called).To(BeTrue())
262 Expect(fn1Called).To(BeTrue())
263 })
264 })
265
266 var _ = Describe("Client timeout", func() {
267 var opt *redis.Options
268 var client *redis.Client
269
270 AfterEach(func() {
271 Expect(client.Close()).NotTo(HaveOccurred())
272 })
273
274 testTimeout := func() {
275 It("Ping timeouts", func() {
276 err := client.Ping().Err()
277 Expect(err).To(HaveOccurred())
278 Expect(err.(net.Error).Timeout()).To(BeTrue())
279 })
280
281 It("Pipeline timeouts", func() {
282 _, err := client.Pipelined(func(pipe redis.Pipeliner) error {
283 pipe.Ping()
284 return nil
285 })
286 Expect(err).To(HaveOccurred())
287 Expect(err.(net.Error).Timeout()).To(BeTrue())
288 })
289
290 It("Subscribe timeouts", func() {
291 if opt.WriteTimeout == 0 {
292 return
293 }
294
295 pubsub := client.Subscribe()
296 defer pubsub.Close()
297
298 err := pubsub.Subscribe("_")
299 Expect(err).To(HaveOccurred())
300 Expect(err.(net.Error).Timeout()).To(BeTrue())
301 })
302
303 It("Tx timeouts", func() {
304 err := client.Watch(func(tx *redis.Tx) error {
305 return tx.Ping().Err()
306 })
307 Expect(err).To(HaveOccurred())
308 Expect(err.(net.Error).Timeout()).To(BeTrue())
309 })
310
311 It("Tx Pipeline timeouts", func() {
312 err := client.Watch(func(tx *redis.Tx) error {
313 _, err := tx.Pipelined(func(pipe redis.Pipeliner) error {
314 pipe.Ping()
315 return nil
316 })
317 return err
318 })
319 Expect(err).To(HaveOccurred())
320 Expect(err.(net.Error).Timeout()).To(BeTrue())
321 })
322 }
323
324 Context("read timeout", func() {
325 BeforeEach(func() {
326 opt = redisOptions()
327 opt.ReadTimeout = time.Nanosecond
328 opt.WriteTimeout = -1
329 client = redis.NewClient(opt)
330 })
331
332 testTimeout()
333 })
334
335 Context("write timeout", func() {
336 BeforeEach(func() {
337 opt = redisOptions()
338 opt.ReadTimeout = -1
339 opt.WriteTimeout = time.Nanosecond
340 client = redis.NewClient(opt)
341 })
342
343 testTimeout()
344 })
345 })
346
347 var _ = Describe("Client OnConnect", func() {
348 var client *redis.Client
349
350 BeforeEach(func() {
351 opt := redisOptions()
352 opt.DB = 0
353 opt.OnConnect = func(cn *redis.Conn) error {
354 return cn.ClientSetName("on_connect").Err()
355 }
356
357 client = redis.NewClient(opt)
358 })
359
360 AfterEach(func() {
361 Expect(client.Close()).NotTo(HaveOccurred())
362 })
363
364 It("calls OnConnect", func() {
365 name, err := client.ClientGetName().Result()
366 Expect(err).NotTo(HaveOccurred())
367 Expect(name).To(Equal("on_connect"))
368 })
369 })
370
View as plain text