func (s *Consul) renewLockSession(initialTTL string, id string, stopRenew chan struct{}) {
	sessionDestroyAttempts := 0
	ttl, err := time.ParseDuration(initialTTL)
	if err != nil {
		return
	}
	go func() {
		for {
			select {
			case <-time.After(ttl / 2):
				entry, _, err := s.client.Session().Renew(id, nil)
				if err != nil {
					continue
				}
				if entry == nil {
					return
				}
				ttl, _ = time.ParseDuration(entry.TTL)
			case <-stopRenew:
				_, err := s.client.Session().Destroy(id, nil)
				if err == nil {
					return
				}
				sessionDestroyAttempts++
				if sessionDestroyAttempts >= MaxSessionDestroyAttempts {
					return
				}
				time.Sleep(ttl / 2)
			}
		}
	}()
}
renewLockSession is used to renew a session Lock, it takes
a stopRenew chan which is used to explicitly stop the session
renew process. The renew routine never stops until a signal is
sent to this channel. If deleting the session fails because the
connection to the store is lost, it keeps trying to delete the
session periodically until it can contact the store, this ensures
that the lock is not maintained indefinitely which ensures liveness
over safety for the lock when the store becomes unavailable.
package consul
import (
	"crypto/tls"
	"errors"
	"net/http"
	"strings"
	"sync"
	"time"
	"github.com/abronan/valkeyrie"
	"github.com/abronan/valkeyrie/store"
	"github.com/hashicorp/consul/api"
)
const (
	DefaultWatchWaitTime = 15 * time.Second
	RenewSessionRetryMax = 5
	MaxSessionDestroyAttempts = 5
	defaultLockTTL = 20 * time.Second
)
var (
	ErrMultipleEndpointsUnsupported = errors.New("consul does not support multiple endpoints")
	ErrSessionRenew = errors.New("cannot set or renew session for ttl, unable to operate on sessions")
)
type Consul struct {
	sync.Mutex
	config *api.Config
	client *api.Client
}
type consulLock struct {
	lock    *api.Lock
	renewCh chan struct{}
}
func Register() {
	valkeyrie.AddStore(store.CONSUL, New)
}
func New(endpoints []string, options *store.Config) (store.Store, error) {
	if len(endpoints) > 1 {
		return nil, ErrMultipleEndpointsUnsupported
	}
	s := &Consul{}
	config := api.DefaultConfig()
	s.config = config
	config.HttpClient = http.DefaultClient
	config.Address = endpoints[0]
	if options != nil {
		if options.TLS != nil {
			s.setTLS(options.TLS)
		}
		if options.ConnectionTimeout != 0 {
			s.setTimeout(options.ConnectionTimeout)
		}
		if options.Token != "" {
			s.setToken(options.Token)
		}
	}
	client, err := api.NewClient(config)
	if err != nil {
		return nil, err
	}
	s.client = client
	return s, nil
}
func (s *Consul) setTLS(tls *tls.Config) {
	s.config.HttpClient.Transport = &http.Transport{
		TLSClientConfig: tls,
	}
	s.config.Scheme = "https"
}
func (s *Consul) setTimeout(time time.Duration) {
	s.config.WaitTime = time
}
func (s *Consul) setToken(token string) {
	s.config.Token = token
}
func (s *Consul) normalize(key string) string {
	key = store.Normalize(key)
	return strings.TrimPrefix(key, "/")
}
func (s *Consul) renewSession(pair *api.KVPair, ttl time.Duration) error {
	session, err := s.getActiveSession(pair.Key)
	if err != nil {
		return err
	}
	if session == "" {
		entry := &api.SessionEntry{
			Behavior:  api.SessionBehaviorDelete,  
			TTL:       (ttl / 2).String(),         
			LockDelay: 1 * time.Millisecond,       
		}
		session, _, err = s.client.Session().Create(entry, nil)
		if err != nil {
			return err
		}
		lockOpts := &api.LockOptions{
			Key:     pair.Key,
			Session: session,
		}
		lock, _ := s.client.LockOpts(lockOpts)
		if lock != nil {
			_, _ = lock.Lock(nil)
		}
	}
	_, _, err = s.client.Session().Renew(session, nil)
	return err
}
func (s *Consul) getActiveSession(key string) (string, error) {
	pair, _, err := s.client.KV().Get(key, nil)
	if err != nil {
		return "", err
	}
	if pair != nil && pair.Session != "" {
		return pair.Session, nil
	}
	return "", nil
}
func (s *Consul) Get(key string, opts *store.ReadOptions) (*store.KVPair, error) {
	options := &api.QueryOptions{
		AllowStale:        false,
		RequireConsistent: true,
	}
	if opts != nil {
		options.RequireConsistent = opts.Consistent
	}
	pair, meta, err := s.client.KV().Get(s.normalize(key), options)
	if err != nil {
		return nil, err
	}
	if pair == nil {
		return nil, store.ErrKeyNotFound
	}
	return &store.KVPair{Key: pair.Key, Value: pair.Value, LastIndex: meta.LastIndex}, nil
}
func (s *Consul) Put(key string, value []byte, opts *store.WriteOptions) error {
	key = s.normalize(key)
	p := &api.KVPair{
		Key:   key,
		Value: value,
		Flags: api.LockFlagValue,
	}
	if opts != nil && opts.TTL > 0 {
		for retry := 1; retry <= RenewSessionRetryMax; retry++ {
			err := s.renewSession(p, opts.TTL)
			if err == nil {
				break
			}
			if retry == RenewSessionRetryMax {
				return ErrSessionRenew
			}
		}
	}
	_, err := s.client.KV().Put(p, nil)
	return err
}
func (s *Consul) Delete(key string) error {
	if _, err := s.Get(key, nil); err != nil {
		return err
	}
	_, err := s.client.KV().Delete(s.normalize(key), nil)
	return err
}
func (s *Consul) Exists(key string, opts *store.ReadOptions) (bool, error) {
	_, err := s.Get(key, opts)
	if err != nil {
		if err == store.ErrKeyNotFound {
			return false, nil
		}
		return false, err
	}
	return true, nil
}
func (s *Consul) List(directory string, opts *store.ReadOptions) ([]*store.KVPair, error) {
	options := &api.QueryOptions{
		AllowStale:        false,
		RequireConsistent: true,
	}
	if opts != nil {
		if !opts.Consistent {
			options.AllowStale = true
			options.RequireConsistent = false
		}
	}
	pairs, _, err := s.client.KV().List(s.normalize(directory), options)
	if err != nil {
		return nil, err
	}
	if len(pairs) == 0 {
		return nil, store.ErrKeyNotFound
	}
	kv := []*store.KVPair{}
	for _, pair := range pairs {
		if pair.Key == directory {
			continue
		}
		kv = append(kv, &store.KVPair{
			Key:       pair.Key,
			Value:     pair.Value,
			LastIndex: pair.ModifyIndex,
		})
	}
	return kv, nil
}
func (s *Consul) DeleteTree(directory string) error {
	if _, err := s.List(directory, nil); err != nil {
		return err
	}
	_, err := s.client.KV().DeleteTree(s.normalize(directory), nil)
	return err
}
func (s *Consul) Watch(key string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan *store.KVPair, error) {
	kv := s.client.KV()
	watchCh := make(chan *store.KVPair)
	go func() {
		defer close(watchCh)
		opts := &api.QueryOptions{WaitTime: DefaultWatchWaitTime}
		for {
			select {
			case <-stopCh:
				return
			default:
			}
			pair, meta, err := kv.Get(key, opts)
			if err != nil {
				return
			}
			if opts.WaitIndex == meta.LastIndex {
				continue
			}
			opts.WaitIndex = meta.LastIndex
			if pair != nil {
				watchCh <- &store.KVPair{
					Key:       pair.Key,
					Value:     pair.Value,
					LastIndex: pair.ModifyIndex,
				}
			}
		}
	}()
	return watchCh, nil
}
func (s *Consul) WatchTree(directory string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan []*store.KVPair, error) {
	kv := s.client.KV()
	watchCh := make(chan []*store.KVPair)
	go func() {
		defer close(watchCh)
		opts := &api.QueryOptions{WaitTime: DefaultWatchWaitTime}
		for {
			select {
			case <-stopCh:
				return
			default:
			}
			pairs, meta, err := kv.List(directory, opts)
			if err != nil {
				return
			}
			if opts.WaitIndex == meta.LastIndex {
				continue
			}
			opts.WaitIndex = meta.LastIndex
			kvpairs := []*store.KVPair{}
			for _, pair := range pairs {
				if pair.Key == directory {
					continue
				}
				kvpairs = append(kvpairs, &store.KVPair{
					Key:       pair.Key,
					Value:     pair.Value,
					LastIndex: pair.ModifyIndex,
				})
			}
			watchCh <- kvpairs
		}
	}()
	return watchCh, nil
}
func (s *Consul) NewLock(key string, options *store.LockOptions) (store.Locker, error) {
	lockOpts := &api.LockOptions{
		Key: s.normalize(key),
	}
	lock := &consulLock{}
	ttl := defaultLockTTL
	if options != nil {
		if options.TTL != 0 {
			ttl = options.TTL
		}
		if options.Value != nil {
			lockOpts.Value = options.Value
		}
	}
	entry := &api.SessionEntry{
		Behavior:  api.SessionBehaviorRelease,  
		TTL:       (ttl / 2).String(),          
		LockDelay: 1 * time.Millisecond,        
	}
	session, _, err := s.client.Session().Create(entry, nil)
	if err != nil {
		return nil, err
	}
	lockOpts.Session = session
	if options != nil {
		lock.renewCh = options.RenewLock
	}
	l, err := s.client.LockOpts(lockOpts)
	if err != nil {
		return nil, err
	}
	if options != nil {
		s.renewLockSession(entry.TTL, session, options.RenewLock)
	}
	lock.lock = l
	return lock, nil
}