Compare commits
3 Commits
Author | SHA1 | Date | |
---|---|---|---|
c8d21b4ef4 | |||
195c604a31 | |||
b263b69031 |
12
options.go
12
options.go
@ -5,14 +5,8 @@ import (
|
||||
"go.unistack.org/micro/v4/options"
|
||||
)
|
||||
|
||||
type configKey struct{}
|
||||
type universalConfigKey struct{}
|
||||
|
||||
func Config(c *redis.Options) options.Option {
|
||||
return options.ContextOption(configKey{}, c)
|
||||
}
|
||||
|
||||
type clusterConfigKey struct{}
|
||||
|
||||
func ClusterConfig(c *redis.ClusterOptions) options.Option {
|
||||
return options.ContextOption(clusterConfigKey{}, c)
|
||||
func UniversalConfig(c *redis.UniversalOptions) options.Option {
|
||||
return options.ContextOption(universalConfigKey{}, c)
|
||||
}
|
||||
|
84
redis.go
84
redis.go
@ -14,21 +14,7 @@ import (
|
||||
|
||||
type Store struct {
|
||||
opts store.Options
|
||||
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
|
||||
cli redis.UniversalClient
|
||||
}
|
||||
|
||||
func (r *Store) Connect(ctx context.Context) error {
|
||||
@ -43,6 +29,10 @@ 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()
|
||||
}
|
||||
@ -346,68 +336,30 @@ func NewStore(opts ...options.Option) *Store {
|
||||
}
|
||||
|
||||
func (r *Store) configure() error {
|
||||
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"}
|
||||
}
|
||||
var redisUniversalOptions *redis.UniversalOptions
|
||||
|
||||
if r.cli != nil && r.opts.Context == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if r.opts.Context != nil {
|
||||
if c, ok := r.opts.Context.Value(configKey{}).(*redis.Options); ok {
|
||||
redisOptions = c
|
||||
if c, ok := r.opts.Context.Value(universalConfigKey{}).(*redis.UniversalOptions); ok {
|
||||
redisUniversalOptions = c
|
||||
if r.opts.TLSConfig != nil {
|
||||
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
|
||||
redisUniversalOptions.TLSConfig = r.opts.TLSConfig
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if redisOptions != nil && redisClusterOptions != nil {
|
||||
return fmt.Errorf("must specify only one option Config or ClusterConfig")
|
||||
}
|
||||
|
||||
if redisOptions == nil && redisClusterOptions == nil && r.cli != nil {
|
||||
if redisUniversalOptions == nil && r.cli != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if redisOptions == nil && redisClusterOptions == nil && len(nodes) == 1 {
|
||||
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,
|
||||
if redisUniversalOptions == nil {
|
||||
redisUniversalOptions = &redis.UniversalOptions{
|
||||
Username: "",
|
||||
Password: "", // no password set
|
||||
DB: 0, // use default DB
|
||||
MaxRetries: 2,
|
||||
MaxRetryBackoff: 256 * time.Millisecond,
|
||||
DialTimeout: 1 * time.Second,
|
||||
@ -419,11 +371,13 @@ func (r *Store) configure() error {
|
||||
}
|
||||
}
|
||||
|
||||
if redisOptions != nil {
|
||||
r.cli = redis.NewClient(redisOptions)
|
||||
} else if redisClusterOptions != nil {
|
||||
r.cli = redis.NewClusterClient(redisClusterOptions)
|
||||
if len(r.opts.Address) > 0 {
|
||||
redisUniversalOptions.Addrs = r.opts.Address
|
||||
} else if len(redisUniversalOptions.Addrs) == 0 {
|
||||
redisUniversalOptions.Addrs = []string{"redis://127.0.0.1:6379"}
|
||||
}
|
||||
|
||||
r.cli = redis.NewUniversalClient(redisUniversalOptions)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
@ -72,11 +72,11 @@ func Test_rkv_configure(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
r := &Store{
|
||||
rc := &Store{
|
||||
opts: tt.fields.options,
|
||||
cli: tt.fields.Client,
|
||||
}
|
||||
err := r.configure()
|
||||
err := rc.configure()
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Errorf("configure() error = %v, wantErr %v", err, tt.wantErr)
|
||||
return
|
||||
|
Loading…
Reference in New Issue
Block a user