func (l *redisLock) tryLock(lockHeld, stopChan chan struct{}) (bool, error) {
	success, item, err := l.redis.AtomicPut(
		l.key,
		l.value,
		l.last,
		&store.WriteOptions{
			TTL: l.ttl,
		})
	if success {
		l.last = item
		go l.holdLock(lockHeld, stopChan)
		return true, nil
	}
	if err != nil && (err == store.ErrKeyNotFound || err == store.ErrKeyModified || err == store.ErrKeyExists) {
		return false, nil
	}
	return false, err
}
tryLock return true, nil when it acquired and hold the lock
and return false, nil when it can't lock now,
and return false, err if any unespected error happened underlying
package redis
import (
	"errors"
	"fmt"
	"log"
	"strings"
	"time"
	"github.com/abronan/valkeyrie"
	"github.com/abronan/valkeyrie/store"
	"gopkg.in/redis.v5"
)
var (
	ErrMultipleEndpointsUnsupported = errors.New("redis: does not support multiple endpoints")
	ErrTLSUnsupported = errors.New("redis does not support tls")
	ErrAbortTryLock = errors.New("redis: lock operation aborted")
)
func Register() {
	valkeyrie.AddStore(store.REDIS, New)
}
func New(endpoints []string, options *store.Config) (store.Store, error) {
	if len(endpoints) > 1 {
		return nil, ErrMultipleEndpointsUnsupported
	}
	if options != nil && options.TLS != nil {
		return nil, ErrTLSUnsupported
	}
	var password string
	if options != nil && options.Password != "" {
		password = options.Password
	}
	return newRedis(endpoints, password, &RawCodec{})
}
func newRedis(endpoints []string, password string, codec Codec) (*Redis, error) {
	client := redis.NewClient(&redis.Options{
		Addr:         endpoints[0],
		DialTimeout:  5 * time.Second,
		ReadTimeout:  30 * time.Second,
		WriteTimeout: 30 * time.Second,
		Password:     password,
	})
	client.ConfigSet("notify-keyspace-events", "KEA")
	var c Codec = &JSONCodec{}
	if codec != nil {
		c = codec
	}
	return &Redis{
		client: client,
		script: redis.NewScript(luaScript()),
		codec:  c,
	}, nil
}
type Redis struct {
	client *redis.Client
	script *redis.Script
	codec  Codec
}
const (
	noExpiration   = time.Duration(0)
	defaultLockTTL = 60 * time.Second
)
func (r *Redis) Put(key string, value []byte, options *store.WriteOptions) error {
	expirationAfter := noExpiration
	if options != nil && options.TTL != 0 {
		expirationAfter = options.TTL
	}
	return r.setTTL(normalize(key), &store.KVPair{
		Key:       key,
		Value:     value,
		LastIndex: sequenceNum(),
	}, expirationAfter)
}
func (r *Redis) setTTL(key string, val *store.KVPair, ttl time.Duration) error {
	valStr, err := r.codec.Encode(val)
	if err != nil {
		return err
	}
	return r.client.Set(key, valStr, ttl).Err()
}
func (r *Redis) Get(key string, opts *store.ReadOptions) (*store.KVPair, error) {
	return r.get(normalize(key))
}
func (r *Redis) get(key string) (*store.KVPair, error) {
	reply, err := r.client.Get(key).Bytes()
	if err != nil {
		if err == redis.Nil {
			return nil, store.ErrKeyNotFound
		}
		return nil, err
	}
	val := store.KVPair{}
	if err := r.codec.Decode(reply, &val); err != nil {
		return nil, err
	}
	if val.Key == "" {
		val.Key = key
	}
	return &val, nil
}
func (r *Redis) Delete(key string) error {
	return r.client.Del(normalize(key)).Err()
}
func (r *Redis) Exists(key string, opts *store.ReadOptions) (bool, error) {
	return r.client.Exists(normalize(key)).Result()
}
func (r *Redis) Watch(key string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan *store.KVPair, error) {
	watchCh := make(chan *store.KVPair)
	nKey := normalize(key)
	get := getter(func() (interface{}, error) {
		pair, err := r.get(nKey)
		if err != nil {
			return nil, err
		}
		return pair, nil
	})
	push := pusher(func(v interface{}) {
		if val, ok := v.(*store.KVPair); ok {
			watchCh <- val
		}
	})
	sub, err := newSubscribe(r.client, regexWatch(nKey, false))
	if err != nil {
		return nil, err
	}
	go func(sub *subscribe, stopCh <-chan struct{}, get getter, push pusher) {
		defer sub.Close()
		msgCh := sub.Receive(stopCh)
		if err := watchLoop(msgCh, stopCh, get, push); err != nil {
			log.Printf("watchLoop in Watch err:%v\n", err)
		}
	}(sub, stopCh, get, push)
	return watchCh, nil
}
func regexWatch(key string, withChildren bool) string {
	var regex string
	if withChildren {
		regex = fmt.Sprintf("__keyspace*:%s*", key)
	} else {
		regex = fmt.Sprintf("__keyspace*:%s", key)
	}
	return regex
}
type getter func() (interface{}, error)
type pusher func(interface{})
func watchLoop(msgCh chan *redis.Message, _ <-chan struct{}, get getter, push pusher) error {
	pair, err := get()
	if err != nil {
		return err
	}
	push(pair)
	for m := range msgCh {
		pair, err := get()
		if err != nil && err != store.ErrKeyNotFound {
			return err
		}
		if err == store.ErrKeyNotFound && (m.Payload == "expire" || m.Payload == "del") {
			push(&store.KVPair{})
		} else {
			push(pair)
		}
	}
	return nil
}
type subscribe struct {
	pubsub  *redis.PubSub
	closeCh chan struct{}
}
func newSubscribe(client *redis.Client, regex string) (*subscribe, error) {
	ch, err := client.PSubscribe(regex)
	if err != nil {
		return nil, err
	}
	return &subscribe{
		pubsub:  ch,
		closeCh: make(chan struct{}),
	}, nil
}
func (s *subscribe) Close() error {
	close(s.closeCh)
	return s.pubsub.Close()
}
func (s *subscribe) Receive(stopCh <-chan struct{}) chan *redis.Message {
	msgCh := make(chan *redis.Message)
	go s.receiveLoop(msgCh, stopCh)
	return msgCh
}
func (s *subscribe) receiveLoop(msgCh chan *redis.Message, stopCh <-chan struct{}) {
	defer close(msgCh)
	for {
		select {
		case <-s.closeCh:
			return
		case <-stopCh:
			return
		default:
			msg, err := s.pubsub.ReceiveMessage()
			if err != nil {
				return
			}
			if msg != nil {
				msgCh <- msg
			}
		}
	}
}
func (r *Redis) WatchTree(directory string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan []*store.KVPair, error) {
	watchCh := make(chan []*store.KVPair)
	nKey := normalize(directory)
	get := getter(func() (interface{}, error) {
		pair, err := r.list(nKey)
		if err != nil {
			return nil, err
		}
		return pair, nil
	})
	push := pusher(func(v interface{}) {
		if _, ok := v.([]*store.KVPair); !ok {
			return
		}
		watchCh <- v.([]*store.KVPair)
	})
	sub, err := newSubscribe(r.client, regexWatch(nKey, true))
	if err != nil {
		return nil, err
	}
	go func(sub *subscribe, stopCh <-chan struct{}, get getter, push pusher) {
		defer sub.Close()
		msgCh := sub.Receive(stopCh)
		if err := watchLoop(msgCh, stopCh, get, push); err != nil {
			log.Printf("watchLoop in WatchTree err:%v\n", err)
		}
	}(sub, stopCh, get, push)
	return watchCh, nil
}
func (r *Redis) NewLock(key string, options *store.LockOptions) (store.Locker, error) {
	var (
		value []byte
		ttl   = defaultLockTTL
	)
	if options != nil && options.TTL != 0 {
		ttl = options.TTL
	}
	if options != nil && len(options.Value) != 0 {
		value = options.Value
	}
	return &redisLock{
		redis:    r,
		last:     nil,
		key:      key,
		value:    value,
		ttl:      ttl,
		unlockCh: make(chan struct{}),
	}, nil
}
type redisLock struct {
	redis    *Redis
	last     *store.KVPair
	unlockCh chan struct{}
	key   string
	value []byte
	ttl   time.Duration
}
func (l *redisLock) Lock(stopCh chan struct{}) (<-chan struct{}, error) {
	lockHeld := make(chan struct{})
	success, err := l.tryLock(lockHeld, stopCh)
	if err != nil {
		return nil, err
	}
	if success {
		return lockHeld, nil
	}
	watch, err := l.redis.Watch(l.key, stopCh, nil)
	if err != nil {
		return nil, err
	}
	for {
		select {
		case <-stopCh:
			return nil, ErrAbortTryLock
		case <-watch:
			success, err := l.tryLock(lockHeld, stopCh)
			if err != nil {
				return nil, err
			}
			if success {
				return lockHeld, nil
			}
		}
	}
}