Compare commits
No commits in common. "master" and "v4.0.0" have entirely different histories.
12
options.go
12
options.go
@ -5,8 +5,14 @@ import (
|
|||||||
"go.unistack.org/micro/v4/options"
|
"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)
|
||||||
}
|
}
|
||||||
|
82
redis.go
82
redis.go
@ -14,7 +14,21 @@ import (
|
|||||||
|
|
||||||
type Store struct {
|
type Store struct {
|
||||||
opts store.Options
|
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 {
|
func (r *Store) Connect(ctx context.Context) error {
|
||||||
@ -29,10 +43,6 @@ func (r *Store) Init(opts ...options.Option) error {
|
|||||||
return r.configure()
|
return r.configure()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *Store) Redis() redis.UniversalClient {
|
|
||||||
return r.cli
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *Store) Disconnect(ctx context.Context) error {
|
func (r *Store) Disconnect(ctx context.Context) error {
|
||||||
return r.cli.Close()
|
return r.cli.Close()
|
||||||
}
|
}
|
||||||
@ -336,27 +346,50 @@ func NewStore(opts ...options.Option) *Store {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r *Store) configure() error {
|
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 {
|
if r.cli != nil && r.opts.Context == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if r.opts.Context != 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 {
|
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
|
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: "",
|
Username: "",
|
||||||
Password: "", // no password set
|
Password: "", // no password set
|
||||||
DB: 0, // use default DB
|
DB: 0, // use default DB
|
||||||
@ -370,14 +403,27 @@ func (r *Store) configure() error {
|
|||||||
TLSConfig: r.opts.TLSConfig,
|
TLSConfig: r.opts.TLSConfig,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} else if redisOptions == nil && redisClusterOptions == nil && len(nodes) > 1 {
|
||||||
if len(r.opts.Address) > 0 {
|
redisClusterOptions = &redis.ClusterOptions{
|
||||||
redisUniversalOptions.Addrs = r.opts.Address
|
Addrs: nodes,
|
||||||
} else if len(redisUniversalOptions.Addrs) == 0 {
|
Username: "",
|
||||||
redisUniversalOptions.Addrs = []string{"redis://127.0.0.1:6379"}
|
Password: "", // no password set
|
||||||
|
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,
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
r.cli = redis.NewUniversalClient(redisUniversalOptions)
|
if redisOptions != nil {
|
||||||
|
r.cli = redis.NewClient(redisOptions)
|
||||||
|
} else if redisClusterOptions != nil {
|
||||||
|
r.cli = redis.NewClusterClient(redisClusterOptions)
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
@ -72,11 +72,11 @@ func Test_rkv_configure(t *testing.T) {
|
|||||||
}
|
}
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
rc := &Store{
|
r := &Store{
|
||||||
opts: tt.fields.options,
|
opts: tt.fields.options,
|
||||||
cli: tt.fields.Client,
|
cli: tt.fields.Client,
|
||||||
}
|
}
|
||||||
err := rc.configure()
|
err := r.configure()
|
||||||
if (err != nil) != tt.wantErr {
|
if (err != nil) != tt.wantErr {
|
||||||
t.Errorf("configure() error = %v, wantErr %v", err, tt.wantErr)
|
t.Errorf("configure() error = %v, wantErr %v", err, tt.wantErr)
|
||||||
return
|
return
|
||||||
|
Loading…
Reference in New Issue
Block a user