feat(redis): allow passing in redis.UniversalClient

Author: nyergler

consumer.go

@@ -53,8 +53,9 @@ type ConsumerOptions struct {
 	// Concurrency dictates how many goroutines to spawn to handle the messages.
 	Concurrency int
 	// RedisClient supersedes the RedisOptions field, and allows you to inject
-	// an already-made *redis.Client for use in the consumer.
-	RedisClient *redis.Client
+	// an already-made Redis Client for use in the consumer. This may be either
+	// the standard client or a cluster client.
+	RedisClient redis.UniversalClient
 	// RedisOptions allows you to configure the underlying Redis connection.
 	// More info here:
 	// https://pkg.go.dev/github.com/go-redis/redis/v7?tab=doc#Options.
@@ -74,7 +75,7 @@ type Consumer struct {
 	Errors chan error
 
 	options   *ConsumerOptions
-	redis     *redis.Client
+	redis     redis.UniversalClient
 	consumers map[string]registeredConsumer
 	streams   []string
 	queue     chan *Message
@@ -121,7 +122,7 @@ func NewConsumerWithOptions(options *ConsumerOptions) (*Consumer, error) {
 		options.ReclaimInterval = 1 * time.Second
 	}
 
-	var r *redis.Client
+	var r redis.UniversalClient
 
 	if options.RedisClient != nil {
 		r = options.RedisClient

producer.go

@@ -20,8 +20,9 @@ type ProducerOptions struct {
 	// manner. More info here: https://redis.io/commands/xadd#capped-streams.
 	ApproximateMaxLength bool
 	// RedisClient supersedes the RedisOptions field, and allows you to inject
-	// an already-made *redis.Client for use in the consumer.
-	RedisClient *redis.Client
+	// an already-made Redis Client for use in the consumer. This may be either
+	// the standard client or a cluster client.
+	RedisClient redis.UniversalClient
 	// RedisOptions allows you to configure the underlying Redis connection.
 	// More info here:
 	// https://pkg.go.dev/github.com/go-redis/redis/v7?tab=doc#Options.
@@ -34,7 +35,7 @@ type ProducerOptions struct {
 // processed later by a Consumer.
 type Producer struct {
 	options *ProducerOptions
-	redis   *redis.Client
+	redis   redis.UniversalClient
 }
 
 var defaultProducerOptions = &ProducerOptions{
@@ -51,7 +52,7 @@ func NewProducer() (*Producer, error) {
 
 // NewProducerWithOptions creates a Producer using custom ProducerOptions.
 func NewProducerWithOptions(options *ProducerOptions) (*Producer, error) {
-	var r *redis.Client
+	var r redis.UniversalClient
 
 	if options.RedisClient != nil {
 		r = options.RedisClient

redis.go

@@ -29,7 +29,7 @@ func newRedisClient(options *RedisOptions) *redis.Client {
 // offers the functionality we need. Specifically, it also that it can connect
 // to the actual instance and that the instance supports Redis streams (i.e.
 // it's at least v5).
-func redisPreflightChecks(client *redis.Client) error {
+func redisPreflightChecks(client redis.UniversalClient) error {
 	info, err := client.Info("server").Result()
 	if err != nil {
 		return err