re
/
redis
forked from mirror/redis
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix race in tests.

This commit is contained in:
Vladimir Mihailenco 2016-04-09 11:34:40 +03:00
parent b351402995
commit 787609d1e6
3 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
export_test.go
View File

@ -14,6 +14,6 @@ func (c *PubSub) Pool() pool.Pooler {
return c.base.connPool return c.base.connPool
} }
func SetReceiveMessageTimeout(d time.Duration) { func (c *PubSub) ReceiveMessageTimeout(timeout time.Duration) (*Message, error) {
receiveMessageTimeout = d return c.receiveMessage(timeout)
} }

8
pubsub.go
View File

@ -8,8 +8,6 @@ import (
"gopkg.in/redis.v3/internal/pool" "gopkg.in/redis.v3/internal/pool"
) )
var receiveMessageTimeout = 5 * time.Second
// Posts a message to the given channel. // Posts a message to the given channel.
func (c *Client) Publish(channel, message string) *IntCmd { func (c *Client) Publish(channel, message string) *IntCmd {
req := NewIntCmd("PUBLISH", channel, message) req := NewIntCmd("PUBLISH", channel, message)
@ -255,9 +253,13 @@ func (c *PubSub) Receive() (interface{}, error) {
// messages. It automatically reconnects to Redis Server and resubscribes // messages. It automatically reconnects to Redis Server and resubscribes
// to channels in case of network errors. // to channels in case of network errors.
func (c *PubSub) ReceiveMessage() (*Message, error) { func (c *PubSub) ReceiveMessage() (*Message, error) {
return c.receiveMessage(5 * time.Second)
}
func (c *PubSub) receiveMessage(timeout time.Duration) (*Message, error) {
var errNum uint var errNum uint
for { for {
msgi, err := c.ReceiveTimeout(receiveMessageTimeout) msgi, err := c.ReceiveTimeout(timeout)
if err != nil { if err != nil {
if !isNetworkError(err) { if !isNetworkError(err) {
return nil, err return nil, err

5
pubsub_test.go
View File

@ -256,8 +256,7 @@ var _ = Describe("PubSub", func() {
}) })
It("should ReceiveMessage after timeout", func() { It("should ReceiveMessage after timeout", func() {
timeout := time.Second timeout := 100 * time.Millisecond
redis.SetReceiveMessageTimeout(timeout)
pubsub, err := client.Subscribe("mychannel") pubsub, err := client.Subscribe("mychannel")
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
@ -276,7 +275,7 @@ var _ = Describe("PubSub", func() {
Expect(n).To(Equal(int64(1))) Expect(n).To(Equal(int64(1)))
}() }()
msg, err := pubsub.ReceiveMessage() msg, err := pubsub.ReceiveMessageTimeout(timeout)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
Expect(msg.Channel).To(Equal("mychannel")) Expect(msg.Channel).To(Equal("mychannel"))
Expect(msg.Payload).To(Equal("hello")) Expect(msg.Payload).To(Equal("hello"))