How to get all sharded pubsub messages?

[thx123]

Hi go-redis experts,

I've been struggling to find the right way to get all sharded pubsub messages from a NewClusterClient. Here is a working example:

ssub.go:

package main

import (
	"context"
	"flag"
	"fmt"
	"os"

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

const logPath = "./ssub.log"

var verbose = flag.Bool("verbose", true, "print info level logs to stdout")

func main() {
	lf, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0660)
	if err != nil {
		logger.Fatalf("Failed to open log file: %v", err)
	}
	defer lf.Close()
	defer logger.Init("SsubLogger", *verbose, true, lf).Close()
	logger.Info("")

	// Create a new RedisCluster client
	rdb := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{"localhost:6379"},
	})

	// Ping the Redis Cluster to ensure connectivity
	if err := rdb.Ping(context.Background()).Err(); err != nil {
		logger.Infof("Could not connect to Redis Cluster: %v\n", err)
	}
	logger.Info("RedisCluster client is ready")

	pubsub := rdb.SSubscribe(context.Background(), "ch0", "chx")
	defer pubsub.Close()
	logger.Info("Subscribed to Redis channels ch0 and chx")
	for i := 1; i <= 10; i++ {
		ch := fmt.Sprintf("ch%d", i)
		pubsub.SSubscribe(context.Background(), ch)
		rdb.SPublish(context.Background(), ch, "DDD PING")
		logger.Info("Subscribed to ", ch)
	}
	chss, _ := rdb.PubSubShardChannels(context.Background(), "*").Result()
	logger.Info("Currently subscribed channels: ", chss)

	for {
		logger.Info("Waiting for sharded Redis pub-sub messages")
		// ReceiveMessage will crash when SSubscribe is used.
		// msg, err := pubsub.ReceiveMessage(context.Background())
		// if err != nil {
		// 	logger.Info(err)
		// }
		msg := <-pubsub.Channel()
		logger.Infof("Received %s %s", msg.Channel, msg.Payload)
	}
}

Output:

go run ssub.go

INFO : 2024/06/06 09:29:50.435537 ssub.go:24:
INFO : 2024/06/06 09:29:50.440662 ssub.go:35: RedisCluster client is ready
INFO : 2024/06/06 09:29:50.441385 ssub.go:39: Subscribed to Redis channels ch0 and chx
INFO : 2024/06/06 09:29:50.442207 ssub.go:44: Subscribed to ch1
INFO : 2024/06/06 09:29:50.442461 ssub.go:44: Subscribed to ch2
INFO : 2024/06/06 09:29:50.442600 ssub.go:44: Subscribed to ch3
INFO : 2024/06/06 09:29:50.442975 ssub.go:44: Subscribed to ch4
INFO : 2024/06/06 09:29:50.443069 ssub.go:44: Subscribed to ch5
INFO : 2024/06/06 09:29:50.443151 ssub.go:44: Subscribed to ch6
INFO : 2024/06/06 09:29:50.443275 ssub.go:44: Subscribed to ch7
INFO : 2024/06/06 09:29:50.443371 ssub.go:44: Subscribed to ch8
INFO : 2024/06/06 09:29:50.443493 ssub.go:44: Subscribed to ch9
INFO : 2024/06/06 09:29:50.443591 ssub.go:44: Subscribed to ch10
INFO : 2024/06/06 09:29:50.443742 ssub.go:47: Currently subscribed channels: [ch8 ch10 ch4]
INFO : 2024/06/06 09:29:50.443777 ssub.go:50: Waiting for sharded Redis pub-sub messages
INFO : 2024/06/06 09:29:50.747216 ssub.go:57: Received ch4 DDD PING
INFO : 2024/06/06 09:29:50.747359 ssub.go:50: Waiting for sharded Redis pub-sub messages
INFO : 2024/06/06 09:29:50.949459 ssub.go:57: Received ch8 DDD PING
INFO : 2024/06/06 09:29:50.949712 ssub.go:50: Waiting for sharded Redis pub-sub messages
INFO : 2024/06/06 09:29:50.949810 ssub.go:57: Received ch10 DDD PING
INFO : 2024/06/06 09:29:50.949885 ssub.go:50: Waiting for sharded Redis pub-sub messages

# comment: only getting sharded messages from one shard while using `redis-cli -c` to `spublish` to all 10 channels (ch1 : ch10)
INFO : 2024/06/06 09:30:09.090908 ssub.go:57: Received ch4 foo
INFO : 2024/06/06 09:30:09.091627 ssub.go:50: Waiting for sharded Redis pub-sub messages

The NewClusterClient is only able to get sharded pubsub messages of multiple channels (ch1 : ch10) from one shard (ch8, ch10 and ch4 in this case), despite the attempt to spublish to all channels to make the NewClusterClient aware of all shards.

There aren't a lot of code examples that I can find, and I've already read related tests (e.g. pubsub_test.go and osscluster_test.go in particular), but still can't figure out the right way. What am I missing? Your pointers would be greatly appreciated!

[thx123]

Is it a bug? Could anyone familiar with the codebase provide some guidance? If you don't have time, some quick pointers to the doc or code would be appreciated.