pubsub.Receive() will timeout

[j178]

#2060 cause SetReadDeadline not called, so PubSub.ReceiveTimeout(ctx, 0) will not get a read deadline set, thus it will receive the i/o timeout error after some time. Related #2139.

Expected Behavior

PubSub.ReceiveTimeout(ctx, 0) will block forever until a message is received, So does PubSub.Receive(), PubSub.ReceiveMessage() and PubSub.Channel().

Current Behavior

PubSub.Receive() will timeout if read nothing for sometime.

received subscribe: hello <nil>
redis: 2022/08/03 22:36:54 pubsub.go:168: redis: discarding bad PubSub connection: read tcp [::1]:53016->[::1]:6379: i/o timeout
received <nil> read tcp [::1]:53016->[::1]:6379: i/o timeout
received subscribe: hello <nil>
redis: 2022/08/03 22:36:57 pubsub.go:168: redis: discarding bad PubSub connection: read tcp [::1]:53035->[::1]:6379: i/o timeout
received <nil> read tcp [::1]:53035->[::1]:6379: i/o timeout
received subscribe: hello <nil>

Possible Solution

Revert #2060

Steps to Reproduce

package main

import (
    "context"
    "fmt"

    "github.com/go-redis/redis/v9"
)

func main() {
    c := redis.NewClient(&redis.Options{})
    pubsub := c.Subscribe(context.Background(), "hello")
    for {
        m, err := pubsub.Receive(context.Background())
        fmt.Println("received", m, err)
    }
}

Context (Environment)

Redis: 7.0.3
go-redis: master

[woutslakhorst]

This is also a problem when using Redis sentinel. Sentinel uses PubSub to synchronise state data. The reconnect behaviour adds log entries every 3 seconds for each sentinel.

[Autumn-27]

The latest version doesn’t seem to be taking effect and this still happens.

redis: discarding bad PubSub connection: read tcp 192.168.124.37:56730->xxxxxx:6379: wsarecv: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.

version: v9 9.5.1
client := redis.NewClient(&redis.Options{
Addr: addr,
Password: password,
DB: db,
ReadTimeout: -1,
})
pubsub := r.client.Subscribe(ctx, channels...)
msg, err := pubsub.ReceiveMessage(ctx)