diff --git a/cluster_test.go b/cluster_test.go index 6788599..58bf738 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -788,8 +788,8 @@ var _ = Describe("ClusterClient timeout", func() { Context("read/write timeout", func() { BeforeEach(func() { opt := redisClusterOptions() - opt.ReadTimeout = 200 * time.Millisecond - opt.WriteTimeout = 200 * time.Millisecond + opt.ReadTimeout = 300 * time.Millisecond + opt.WriteTimeout = 300 * time.Millisecond opt.MaxRedirects = 1 client = cluster.clusterClient(opt) diff --git a/main_test.go b/main_test.go index 7c5a6a9..4eddc1e 100644 --- a/main_test.go +++ b/main_test.go @@ -142,7 +142,7 @@ func redisRingOptions() *redis.RingOptions { } } -func perform(n int, cbs ...func(int)) { +func performAsync(n int, cbs ...func(int)) *sync.WaitGroup { var wg sync.WaitGroup for _, cb := range cbs { for i := 0; i < n; i++ { @@ -155,6 +155,11 @@ func perform(n int, cbs ...func(int)) { }(cb, i) } } + return &wg +} + +func perform(n int, cbs ...func(int)) { + wg := performAsync(n, cbs...) wg.Wait() } diff --git a/race_test.go b/race_test.go index 05f338c..789c8ba 100644 --- a/race_test.go +++ b/race_test.go @@ -5,6 +5,7 @@ import ( "fmt" "net" "strconv" + "sync/atomic" "testing" "time" @@ -258,6 +259,30 @@ var _ = Describe("races", func() { Expect(err).NotTo(HaveOccurred()) Expect(n).To(Equal(int64(N))) }) + + It("should BLPop", func() { + var received uint32 + wg := performAsync(C, func(id int) { + for { + v, err := client.BLPop(3*time.Second, "list").Result() + if err != nil { + break + } + Expect(v).To(Equal([]string{"list", "hello"})) + atomic.AddUint32(&received, 1) + } + }) + + perform(C, func(id int) { + for i := 0; i < N; i++ { + err := client.LPush("list", "hello").Err() + Expect(err).NotTo(HaveOccurred()) + } + }) + + wg.Wait() + Expect(received).To(Equal(uint32(C * N))) + }) }) func bigVal() []byte {