func (r *Redis) Lock(key string, options ...store.LockOption) (store.Locker, error) {
	var (
		value []byte
		ttl   = defaultLockTTL
	)
	lOption := &store.LockOptions{}
	for _, option := range options {
		option(lOption)
	}
	if lOption.TTL != noExpiration {
		ttl = lOption.TTL
	}
	if len(lOption.Value) != 0 {
		value = lOption.Value
	}
	return &redisLock{
		redis:    r,
		last:     nil,
		key:      key,
		value:    value,
		ttl:      ttl,
		unlockCh: make(chan struct{}),
	}, nil
}
NewLock creates a lock for a given key.
The returned Locker is not held and must be acquired
with `.Lock`. The Value is optional.
package redis
import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"
	store "github.com/getamis/sirius/kv"
	"github.com/getamis/sirius/log"
	"gopkg.in/redis.v5"
)
var (
	ErrMultipleEndpointsUnsupported = errors.New("redis does not support multiple endpoints")
	ErrAbortTryLock = errors.New("redis: lock operation aborted")
)
func New(endpoints []string, notification bool, options ...RedisOption) (*Redis, error) {
	if len(endpoints) > 1 {
		return nil, ErrMultipleEndpointsUnsupported
	}
	return new(endpoints, notification, options...), nil
}
func new(endpoints []string, notification bool, options ...RedisOption) *Redis {
	rOption := &redis.Options{
		Addr:         endpoints[0],
		Password:     "",  
		DialTimeout:  5 * time.Second,
		ReadTimeout:  30 * time.Second,
		WriteTimeout: 30 * time.Second,
	}
	for _, option := range options {
		option(rOption)
	}
	client := redis.NewClient(rOption)
	if notification {
		client.ConfigSet("notify-keyspace-events", "KA")
	}
	return &Redis{
		client: client,
		script: redis.NewScript(luaScript()),
		codec:  defaultCodec{},
	}
}
type defaultCodec struct{}
func (c defaultCodec) encode(kv *store.KeyValue) (string, error) {
	b, err := json.Marshal(kv)
	return string(b), err
}
func (c defaultCodec) decode(b string, kv *store.KeyValue) error {
	return json.Unmarshal([]byte(b), kv)
}
type Redis struct {
	client *redis.Client
	script *redis.Script
	codec  defaultCodec
}
const (
	noExpiration   = time.Duration(0)
	defaultLockTTL = 60 * time.Second
)
func (r *Redis) Put(key string, value []byte, options ...store.PutOption) error {
	pOption := &store.PutOptions{}
	for _, option := range options {
		option(pOption)
	}
	expirationAfter := noExpiration
	if pOption.TTL != noExpiration {
		expirationAfter = pOption.TTL
	}
	return r.setTTL(normalize(key), &store.KeyValue{
		Key:      key,
		Value:    value,
		Revision: sequenceNum(),
	}, expirationAfter)
}
func (r *Redis) setTTL(key string, val *store.KeyValue, 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) (*store.KeyValue, error) {
	return r.get(normalize(key))
}
func (r *Redis) get(key string) (*store.KeyValue, error) {
	reply, err := r.client.Get(key).Bytes()
	if err != nil {
		if err == redis.Nil {
			return nil, store.ErrKeyNotFound
		}
		return nil, err
	}
	val := store.KeyValue{}
	if err := r.codec.decode(string(reply), &val); err != nil {
		return nil, err
	}
	return &val, nil
}
func (r *Redis) Delete(key string) error {
	return r.client.Del(normalize(key)).Err()
}
func (r *Redis) Exists(key string) (bool, error) {
	return r.client.Exists(normalize(key)).Result()
}
func (r *Redis) Watch(key string, stopCh <-chan struct{}) (<-chan *store.KeyValue, error) {
	watchCh := make(chan *store.KeyValue)
	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.KeyValue); 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()
		defer close(watchCh)
		msgCh := sub.Receive(stopCh)
		if err := watchLoop(msgCh, stopCh, get, push); err != nil {
			log.Info("watchLoop in Watch", "err", 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, stopCh <-chan struct{}, get getter, push pusher) error {
	pair, err := get()
	if err != nil {
		return err
	}
	push(pair)
	for range msgCh {
		pair, err := get()
		if err != nil {
			return err
		}
		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{}) (<-chan []*store.KeyValue, error) {
	watchCh := make(chan []*store.KeyValue)
	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.KeyValue); !ok {
			return
		}
		watchCh <- v.([]*store.KeyValue)
	})
	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()
		defer close(watchCh)
		msgCh := sub.Receive(stopCh)
		if err := watchLoop(msgCh, stopCh, get, push); err != nil {
			log.Info("watchLoop in WatchTree", "err", err)
		}
	}(sub, stopCh, get, push)
	return watchCh, nil
}