Merge pull request #658 from go-redis/fix/chan-once

Create PubSub channel once

pubsub.go

@@ -29,6 +29,9 @@ type PubSub struct {
 	closed   bool
 
 	cmd *Cmd
+
+	chOnce sync.Once
+	ch     chan *Message
 }
 
 func (c *PubSub) conn() (*pool.Conn, error) {
@@ -346,24 +349,27 @@ func (c *PubSub) receiveMessage(timeout time.Duration) (*Message, error) {
 	}
 }
 
-// Channel returns a channel for concurrently receiving messages.
+// Channel returns a Go channel for concurrently receiving messages.
-// The channel is closed with PubSub.
+// The channel is closed with PubSub. Receive or ReceiveMessage APIs
+// can not be used after channel is created.
 func (c *PubSub) Channel() <-chan *Message {
-	ch := make(chan *Message, 100)
+	c.chOnce.Do(func() {
-	go func() {
+		c.ch = make(chan *Message, 100)
-		for {
+		go func() {
-			msg, err := c.ReceiveMessage()
+			for {
-			if err != nil {
+				msg, err := c.ReceiveMessage()
-				if err == pool.ErrClosed {
+				if err != nil {
-					break
+					if err == pool.ErrClosed {
+						break
+					}
+					continue
 				}
-				continue
+				c.ch <- msg
 			}
-			ch <- msg
+			close(c.ch)
-		}
+		}()
-		close(ch)
+	})
-	}()
+	return c.ch
-	return ch
 }
 
 func appendIfNotExists(ss []string, es ...string) []string {