func (r *Redis) mget(directory string, keys ...string) ([]*store.KVPair, error) {
	replies, err := r.client.MGet(keys...).Result()
	if err != nil {
		return nil, err
	}
	pairs := []*store.KVPair{}
	for i, reply := range replies {
		var sreply string
		if _, ok := reply.(string); ok {
			sreply = reply.(string)
		}
		if sreply == "" {
			continue
		}
		pair := &store.KVPair{}
		if err := r.codec.Decode([]byte(sreply), pair); err != nil {
			return nil, err
		}
		if pair.Key == "" {
			pair.Key = keys[i]
		}
		if normalize(pair.Key) != directory {
			pairs = append(pairs, pair)
		}
	}
	return pairs, nil
}
mget values given their keys
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
			}
		}
	}
}
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
}
func (l *redisLock) holdLock(lockHeld, stopChan chan struct{}) {
	defer close(lockHeld)
	hold := func() error {
		_, item, err := l.redis.AtomicPut(
			l.key,
			l.value,
			l.last,
			&store.WriteOptions{
				TTL: l.ttl,
			})
		if err == nil {
			l.last = item
		}
		return err
	}
	heartbeat := time.NewTicker(l.ttl / 3)
	defer heartbeat.Stop()
	for {
		select {
		case <-heartbeat.C:
			if err := hold(); err != nil {
				return
			}
		case <-l.unlockCh:
			return
		case <-stopChan:
			return
		}
	}
}
func (l *redisLock) Unlock() error {
	l.unlockCh <- struct{}{}
	_, err := l.redis.AtomicDelete(l.key, l.last)
	if err != nil {
		return err
	}
	l.last = nil
	return err
}
func (r *Redis) List(directory string, opts *store.ReadOptions) ([]*store.KVPair, error) {
	return r.list(normalize(directory))
}
func (r *Redis) list(directory string) ([]*store.KVPair, error) {
	var allKeys []string
	regex := scanRegex(directory)  
	allKeys, err := r.keys(regex)
	if err != nil {
		return nil, err
	}
	return r.mget(directory, allKeys...)
}
func (r *Redis) keys(regex string) ([]string, error) {
	const (
		startCursor  = 0
		endCursor    = 0
		defaultCount = 10
	)
	var allKeys []string
	keys, nextCursor, err := r.client.Scan(startCursor, regex, defaultCount).Result()
	if err != nil {
		return nil, err
	}
	allKeys = append(allKeys, keys...)
	for nextCursor != endCursor {
		keys, nextCursor, err = r.client.Scan(nextCursor, regex, defaultCount).Result()
		if err != nil {
			return nil, err
		}
		allKeys = append(allKeys, keys...)
	}
	if len(allKeys) == 0 {
		return nil, store.ErrKeyNotFound
	}
	return allKeys, nil
}