Merge pull request #1463 from go-redis/fix/sub-renewal

Faster renew the subscription

internal/pool/pool.go

@@ -163,6 +163,7 @@ func (p *ConnPool) newConn(ctx context.Context, pooled bool) (*Conn, error) {
 		}
 	}
 	p.connsMu.Unlock()
+
 	return cn, nil
 }
 
@@ -408,8 +409,10 @@ func (p *ConnPool) closed() bool {
 }
 
 func (p *ConnPool) Filter(fn func(*Conn) bool) error {
-	var firstErr error
 	p.connsMu.Lock()
+	defer p.connsMu.Unlock()
+
+	var firstErr error
 	for _, cn := range p.conns {
 		if fn(cn) {
 			if err := p.closeConn(cn); err != nil && firstErr == nil {
@@ -417,7 +420,6 @@ func (p *ConnPool) Filter(fn func(*Conn) bool) error {
 			}
 		}
 	}
-	p.connsMu.Unlock()
 	return firstErr
 }
 

pubsub.go

@@ -13,7 +13,10 @@ import (
 	"github.com/go-redis/redis/v8/internal/proto"
 )
 
-const pingTimeout = 30 * time.Second
+const (
+	pingTimeout     = time.Second
+	chanSendTimeout = time.Minute
+)
 
 var errPingTimeout = errors.New("redis: ping timeout")
 
@@ -454,7 +457,6 @@ func (c *PubSub) getContext() context.Context {
 	if c.cmd != nil {
 		return c.cmd.ctx
 	}
-
 	return context.Background()
 }
 
@@ -462,7 +464,7 @@ func (c *PubSub) initPing() {
 	ctx := context.TODO()
 	c.ping = make(chan struct{}, 1)
 	go func() {
-		timer := time.NewTimer(pingTimeout)
+		timer := time.NewTimer(time.Minute)
 		timer.Stop()
 
 		healthy := true
@@ -499,7 +501,7 @@ func (c *PubSub) initMsgChan(size int) {
 	ctx := context.TODO()
 	c.msgCh = make(chan *Message, size)
 	go func() {
-		timer := time.NewTimer(pingTimeout)
+		timer := time.NewTimer(time.Minute)
 		timer.Stop()
 
 		var errCount int
@@ -531,7 +533,7 @@ func (c *PubSub) initMsgChan(size int) {
 			case *Pong:
 				// Ignore.
 			case *Message:
-				timer.Reset(pingTimeout)
+				timer.Reset(chanSendTimeout)
 				select {
 				case c.msgCh <- msg:
 					if !timer.Stop() {
@@ -540,7 +542,10 @@ func (c *PubSub) initMsgChan(size int) {
 				case <-timer.C:
 					internal.Logger.Printf(
 						c.getContext(),
-						"redis: %s channel is full for %s (message is dropped)", c, pingTimeout)
+						"redis: %s channel is full for %s (message is dropped)",
+						c,
+						chanSendTimeout,
+					)
 				}
 			default:
 				internal.Logger.Printf(c.getContext(), "redis: unknown message type: %T", msg)

sentinel_test.go

@@ -70,12 +70,12 @@ var _ = Describe("Sentinel", func() {
 			return client.Get(ctx, "foo").Err()
 		}, "15s", "100ms").ShouldNot(HaveOccurred())
 
-		// Publish message to check if subscription is renewed.
-		err = client.Publish(ctx, "foo", "hello").Err()
-		Expect(err).NotTo(HaveOccurred())
-
+		// Check if subscription is renewed.
 		var msg *redis.Message
-		Eventually(ch, "15s").Should(Receive(&msg))
+		Eventually(func() <-chan *redis.Message {
+			_ = client.Publish(ctx, "foo", "hello").Err()
+			return ch
+		}, "15s").Should(Receive(&msg))
 		Expect(msg.Channel).To(Equal("foo"))
 		Expect(msg.Payload).To(Equal("hello"))
 	})