func (s *Zookeeper) getListWithPath(path string, keys []string, opts *store.ReadOptions) ([]*store.KVPair, error) {
	kvs := []*store.KVPair{}
	for _, key := range keys {
		pair, err := s.Get(strings.TrimSuffix(path, "/")+s.normalize(key), opts)
		if err != nil {
			return nil, err
		}
		kvs = append(kvs, &store.KVPair{
			Key:       key,
			Value:     pair.Value,
			LastIndex: pair.LastIndex,
		})
	}
	return kvs, nil
}
getListWithPath gets the key/value pairs for a list of keys under
a given path.

This is generally used when we get a list of child keys which
are stripped out of their path (for example when using ChildrenW).
package zookeeper
import (
	"strings"
	"time"
	"github.com/abronan/valkeyrie"
	"github.com/abronan/valkeyrie/store"
	"github.com/samuel/go-zookeeper/zk"
)
const (
	SOH = "\x01"
	defaultTimeout = 10 * time.Second
	syncRetryLimit = 5
)
type Zookeeper struct {
	timeout time.Duration
	client  *zk.Conn
}
type zookeeperLock struct {
	client *zk.Conn
	lock   *zk.Lock
	key    string
	value  []byte
}
func Register() {
	valkeyrie.AddStore(store.ZK, New)
}
func New(endpoints []string, options *store.Config) (store.Store, error) {
	s := &Zookeeper{}
	s.timeout = defaultTimeout
	if options != nil {
		if options.ConnectionTimeout != 0 {
			s.setTimeout(options.ConnectionTimeout)
		}
	}
	conn, _, err := zk.Connect(endpoints, s.timeout)
	if err != nil {
		return nil, err
	}
	s.client = conn
	return s, nil
}
func (s *Zookeeper) setTimeout(time time.Duration) {
	s.timeout = time
}
func (s *Zookeeper) Get(key string, opts *store.ReadOptions) (pair *store.KVPair, err error) {
	resp, meta, err := s.get(key)
	if err != nil {
		return nil, err
	}
	pair = &store.KVPair{
		Key:       key,
		Value:     resp,
		LastIndex: uint64(meta.Version),
	}
	return pair, nil
}
func (s *Zookeeper) createFullPath(path []string, data []byte, ephemeral bool) error {
	for i := 1; i <= len(path); i++ {
		newpath := "/" + strings.Join(path[:i], "/")
		if i == len(path) {
			flag := 0
			if ephemeral {
				flag = zk.FlagEphemeral
			}
			_, err := s.client.Create(newpath, data, int32(flag), zk.WorldACL(zk.PermAll))
			return err
		}
		_, err := s.client.Create(newpath, []byte{}, 0, zk.WorldACL(zk.PermAll))
		if err != nil {
			if err != zk.ErrNodeExists {
				return err
			}
		}
	}
	return nil
}
func (s *Zookeeper) Put(key string, value []byte, opts *store.WriteOptions) error {
	fkey := s.normalize(key)
	exists, err := s.Exists(key, nil)
	if err != nil {
		return err
	}
	if !exists {
		if opts != nil && opts.TTL > 0 {
			_ = s.createFullPath(store.SplitKey(strings.TrimSuffix(key, "/")), value, true)
		} else {
			_ = s.createFullPath(store.SplitKey(strings.TrimSuffix(key, "/")), value, false)
		}
	} else {
		_, err = s.client.Set(fkey, value, -1)
	}
	return err
}
func (s *Zookeeper) Delete(key string) error {
	err := s.client.Delete(s.normalize(key), -1)
	if err == zk.ErrNoNode {
		return store.ErrKeyNotFound
	}
	return err
}
func (s *Zookeeper) Exists(key string, opts *store.ReadOptions) (bool, error) {
	exists, _, err := s.client.Exists(s.normalize(key))
	if err != nil {
		return false, err
	}
	return exists, nil
}
func (s *Zookeeper) Watch(key string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan *store.KVPair, error) {
	watchCh := make(chan *store.KVPair)
	go func() {
		defer close(watchCh)
		var fireEvt = true
		for {
			resp, meta, eventCh, err := s.getW(key)
			if err != nil {
				return
			}
			if fireEvt {
				watchCh <- &store.KVPair{
					Key:       key,
					Value:     resp,
					LastIndex: uint64(meta.Version),
				}
			}
			select {
			case e := <-eventCh:
				fireEvt = e.Type == zk.EventNodeDataChanged
			case <-stopCh:
				return
			}
		}
	}()
	return watchCh, nil
}
func (s *Zookeeper) WatchTree(directory string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan []*store.KVPair, error) {
	watchCh := make(chan []*store.KVPair)
	go func() {
		defer close(watchCh)
		var fireEvt = true
		for {
		WATCH:
			keys, _, eventCh, err := s.client.ChildrenW(s.normalize(directory))
			if err != nil {
				return
			}
			if fireEvt {
				kvs, err := s.getListWithPath(directory, keys, opts)
				if err != nil {
					goto WATCH
				}
				watchCh <- kvs
			}
			select {
			case e := <-eventCh:
				fireEvt = e.Type == zk.EventNodeChildrenChanged
			case <-stopCh:
				return
			}
		}
	}()
	return watchCh, nil
}
func (s *Zookeeper) listChildren(directory string) ([]string, error) {
	children, _, err := s.client.Children(s.normalize(directory))
	if err != nil {
		if err == zk.ErrNoNode {
			return nil, store.ErrKeyNotFound
		}
		return nil, err
	}
	return children, nil
}
func (s *Zookeeper) listChildrenRecursive(list *[]string, directory string) error {
	children, err := s.listChildren(directory)
	if err != nil {
		return err
	}
	if len(children) == 0 {
		return nil
	}
	for _, c := range children {
		c = strings.TrimSuffix(directory, "/") + "/" + c
		err := s.listChildrenRecursive(list, c)
		if err != nil && err != zk.ErrNoChildrenForEphemerals {
			return err
		}
		*list = append(*list, c)
	}
	return nil
}
func (s *Zookeeper) List(directory string, opts *store.ReadOptions) ([]*store.KVPair, error) {
	children := make([]string, 0)
	err := s.listChildrenRecursive(&children, directory)
	if err != nil {
		return nil, err
	}
	kvs, err := s.getList(children, opts)
	if err != nil {
		if err == store.ErrKeyNotFound {
			return s.List(directory, opts)
		}
		return nil, err
	}
	return kvs, nil
}
func (s *Zookeeper) DeleteTree(directory string) error {
	children, err := s.listChildren(directory)
	if err != nil {
		return err
	}
	var reqs []interface{}
	for _, c := range children {
		reqs = append(reqs, &zk.DeleteRequest{
			Path:    s.normalize(directory + "/" + c),
			Version: -1,
		})
	}
	_, err = s.client.Multi(reqs...)
	return err
}
func (s *Zookeeper) AtomicPut(key string, value []byte, previous *store.KVPair, _ *store.WriteOptions) (bool, *store.KVPair, error) {
	var lastIndex uint64
	if previous != nil {
		meta, err := s.client.Set(s.normalize(key), value, int32(previous.LastIndex))
		if err != nil {
			if err == zk.ErrBadVersion {
				return false, nil, store.ErrKeyModified
			}
			return false, nil, err
		}
		lastIndex = uint64(meta.Version)
	} else {
		_, err := s.client.Create(s.normalize(key), value, 0, zk.WorldACL(zk.PermAll))
		if err != nil {
			if err == zk.ErrNoNode {
				parts := store.SplitKey(strings.TrimSuffix(key, "/"))
				parts = parts[:len(parts)-1]
				if err = s.createFullPath(parts, []byte{}, false); err != nil {
					return false, nil, err
				}
				if _, err := s.client.Create(s.normalize(key), value, 0, zk.WorldACL(zk.PermAll)); err != nil {
					if err == zk.ErrNodeExists {
						return false, nil, store.ErrKeyExists
					}
					return false, nil, err
				}
			} else {
				if err == zk.ErrNodeExists {
					return false, nil, store.ErrKeyExists
				}
				return false, nil, err
			}
		}
		lastIndex = 0  
	}
	pair := &store.KVPair{
		Key:       key,
		Value:     value,
		LastIndex: lastIndex,
	}
	return true, pair, nil
}
func (s *Zookeeper) AtomicDelete(key string, previous *store.KVPair) (bool, error) {
	if previous == nil {
		return false, store.ErrPreviousNotSpecified
	}
	err := s.client.Delete(s.normalize(key), int32(previous.LastIndex))
	if err != nil {
		if err == zk.ErrNoNode {
			return false, store.ErrKeyNotFound
		}
		if err == zk.ErrBadVersion {
			return false, store.ErrKeyModified
		}
		return false, err
	}
	return true, nil
}
func (s *Zookeeper) NewLock(key string, options *store.LockOptions) (lock store.Locker, err error) {
	value := []byte("")
	if options != nil {
		if options.Value != nil {
			value = options.Value
		}
	}
	lock = &zookeeperLock{
		client: s.client,
		key:    s.normalize(key),
		value:  value,
		lock:   zk.NewLock(s.client, s.normalize(key), zk.WorldACL(zk.PermAll)),
	}
	return lock, err
}
func (l *zookeeperLock) Lock(stopChan chan struct{}) (<-chan struct{}, error) {
	err := l.lock.Lock()
	lostCh := make(chan struct{})
	if err == nil {
		_, err = l.client.Set(l.key, l.value, -1)
		if err == nil {
			go l.monitorLock(stopChan, lostCh)
		}
	}
	return lostCh, err
}
func (l *zookeeperLock) Unlock() error {
	return l.lock.Unlock()
}
func (s *Zookeeper) Close() {
	s.client.Close()
}
func (s *Zookeeper) normalize(key string) string {
	key = store.Normalize(key)
	return strings.TrimSuffix(key, "/")
}
func (l *zookeeperLock) monitorLock(stopCh <-chan struct{}, lostCh chan struct{}) {
	defer close(lostCh)
	for {
		_, _, eventCh, err := l.client.GetW(l.key)
		if err != nil {
			return
		}
		select {
		case e := <-eventCh:
			if e.Type == zk.EventNotWatching ||
				(e.Type == zk.EventSession && e.State == zk.StateExpired) {
				return
			} else if e.Type == zk.EventNodeDataChanged {
				return
			}
		case <-stopCh:
			return
		}
	}
}
func (s *Zookeeper) get(key string) ([]byte, *zk.Stat, error) {
	var resp []byte
	var meta *zk.Stat
	var err error
	for i := 0; i <= syncRetryLimit; i++ {
		resp, meta, err = s.client.Get(s.normalize(key))
		if err != nil {
			if err == zk.ErrNoNode {
				return nil, nil, store.ErrKeyNotFound
			}
			return nil, nil, err
		}
		if string(resp) != SOH && string(resp) != "" {
			return resp, meta, nil
		}
		if i < syncRetryLimit {
			if _, err = s.client.Sync(s.normalize(key)); err != nil {
				return nil, nil, err
			}
		}
	}
	return resp, meta, nil
}
func (s *Zookeeper) getW(key string) ([]byte, *zk.Stat, <-chan zk.Event, error) {
	var resp []byte
	var meta *zk.Stat
	var ech <-chan zk.Event
	var err error
	for i := 0; i <= syncRetryLimit; i++ {
		resp, meta, ech, err = s.client.GetW(s.normalize(key))
		if err != nil {
			if err == zk.ErrNoNode {
				return nil, nil, nil, store.ErrKeyNotFound
			}
			return nil, nil, nil, err
		}
		if string(resp) != SOH && string(resp) != "" {
			return resp, meta, ech, nil
		}
		if i < syncRetryLimit {
			if _, err = s.client.Sync(s.normalize(key)); err != nil {
				return nil, nil, nil, err
			}
		}
	}
	return resp, meta, ech, nil
}