options.go

@@ -5,8 +5,14 @@ import (
 	"go.unistack.org/micro/v4/options"
 )
 
-type universalConfigKey struct{}
+type configKey struct{}
 
-func UniversalConfig(c *redis.UniversalOptions) options.Option {
+func Config(c *redis.Options) options.Option {
-	return options.ContextOption(universalConfigKey{}, c)
+	return options.ContextOption(configKey{}, c)
+}
+
+type clusterConfigKey struct{}
+
+func ClusterConfig(c *redis.ClusterOptions) options.Option {
+	return options.ContextOption(clusterConfigKey{}, c)
 }

redis.go

@@ -14,7 +14,21 @@ import (
 
 type Store struct {
 	opts store.Options
-	cli  redis.UniversalClient
+	cli  redisClient
+}
+
+type redisClient interface {
+	Get(ctx context.Context, key string) *redis.StringCmd
+	Del(ctx context.Context, keys ...string) *redis.IntCmd
+	Set(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.StatusCmd
+	Keys(ctx context.Context, pattern string) *redis.StringSliceCmd
+	MGet(ctx context.Context, keys ...string) *redis.SliceCmd
+	MSet(ctx context.Context, kv ...interface{}) *redis.StatusCmd
+	Exists(ctx context.Context, keys ...string) *redis.IntCmd
+	Ping(ctx context.Context) *redis.StatusCmd
+	Pipeline() redis.Pipeliner
+	Pipelined(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error)
+	Close() error
 }
 
 func (r *Store) Connect(ctx context.Context) error {
@@ -29,10 +43,6 @@ func (r *Store) Init(opts ...options.Option) error {
 	return r.configure()
 }
 
-func (r *Store) Redis() redis.UniversalClient {
-	return r.cli
-}
-
 func (r *Store) Disconnect(ctx context.Context) error {
 	return r.cli.Close()
 }
@@ -336,30 +346,68 @@ func NewStore(opts ...options.Option) *Store {
 }
 
 func (r *Store) configure() error {
-	var redisUniversalOptions *redis.UniversalOptions
+	var redisOptions *redis.Options
+	var redisClusterOptions *redis.ClusterOptions
+	var err error
+
+	nodes := r.opts.Address
+
+	if len(nodes) == 0 {
+		nodes = []string{"redis://127.0.0.1:6379"}
+	}
 
 	if r.cli != nil && r.opts.Context == nil {
 		return nil
 	}
 
 	if r.opts.Context != nil {
-		if c, ok := r.opts.Context.Value(universalConfigKey{}).(*redis.UniversalOptions); ok {
+		if c, ok := r.opts.Context.Value(configKey{}).(*redis.Options); ok {
-			redisUniversalOptions = c
+			redisOptions = c
 			if r.opts.TLSConfig != nil {
-				redisUniversalOptions.TLSConfig = r.opts.TLSConfig
+				redisOptions.TLSConfig = r.opts.TLSConfig
+			}
+		}
+
+		if c, ok := r.opts.Context.Value(clusterConfigKey{}).(*redis.ClusterOptions); ok {
+			redisClusterOptions = c
+			if r.opts.TLSConfig != nil {
+				redisClusterOptions.TLSConfig = r.opts.TLSConfig
 			}
 		}
 	}
 
-	if redisUniversalOptions == nil && r.cli != nil {
+	if redisOptions != nil && redisClusterOptions != nil {
+		return fmt.Errorf("must specify only one option Config or ClusterConfig")
+	}
+
+	if redisOptions == nil && redisClusterOptions == nil && r.cli != nil {
 		return nil
 	}
 
-	if redisUniversalOptions == nil {
+	if redisOptions == nil && redisClusterOptions == nil && len(nodes) == 1 {
-		redisUniversalOptions = &redis.UniversalOptions{
+		redisOptions, err = redis.ParseURL(nodes[0])
+		if err != nil {
+			// Backwards compatibility
+			redisOptions = &redis.Options{
+				Addr:            nodes[0],
+				Username:        "",
+				Password:        "", // no password set
+				DB:              0,  // use default DB
+				MaxRetries:      2,
+				MaxRetryBackoff: 256 * time.Millisecond,
+				DialTimeout:     1 * time.Second,
+				ReadTimeout:     1 * time.Second,
+				WriteTimeout:    1 * time.Second,
+				PoolTimeout:     1 * time.Second,
+				MinIdleConns:    10,
+				TLSConfig:       r.opts.TLSConfig,
+			}
+		}
+	} else if redisOptions == nil && redisClusterOptions == nil && len(nodes) > 1 {
+		redisClusterOptions = &redis.ClusterOptions{
+			Addrs:           nodes,
 			Username:        "",
 			Password:        "", // no password set
-			DB:              0,  // use default DB
 			MaxRetries:      2,
 			MaxRetryBackoff: 256 * time.Millisecond,
 			DialTimeout:     1 * time.Second,
@@ -371,13 +419,11 @@ func (r *Store) configure() error {
 		}
 	}
 
-	if len(r.opts.Address) > 0 {
+	if redisOptions != nil {
-		redisUniversalOptions.Addrs = r.opts.Address
+		r.cli = redis.NewClient(redisOptions)
-	} else if len(redisUniversalOptions.Addrs) == 0 {
+	} else if redisClusterOptions != nil {
-		redisUniversalOptions.Addrs = []string{"redis://127.0.0.1:6379"}
+		r.cli = redis.NewClusterClient(redisClusterOptions)
 	}
 
-	r.cli = redis.NewUniversalClient(redisUniversalOptions)
-
 	return nil
 }

redis_test.go

@@ -72,11 +72,11 @@ func Test_rkv_configure(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			rc := &Store{
+			r := &Store{
 				opts: tt.fields.options,
 				cli:  tt.fields.Client,
 			}
-			err := rc.configure()
+			err := r.configure()
 			if (err != nil) != tt.wantErr {
 				t.Errorf("configure() error = %v, wantErr %v", err, tt.wantErr)
 				return