func (self *NsqLookupdEtcdMgr) watchTopics() {
	atomic.StoreInt32(&self.ifTopicChanged, 1)
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		select {
		case <-self.watchTopicsStopCh:
			coordLog.Infof("watch key[%s] will be stopped.", self.topicRoot)
			cancel()
		}
	}()
	watchWaitAndDo(ctx, self.client, self.topicRoot, true, func(rsp *client.Response) {
		mi := getMaxIndexFromWatchRsp(rsp)
		coordLog.Infof("topic max changed from %v to: %v, cluster-index: %v",
			atomic.LoadUint64(&self.watchedClusterIndex), mi, rsp.Index)
		if mi > 0 {
			atomic.StoreUint64(&self.watchedClusterIndex, mi)
		}
		atomic.StoreInt32(&self.ifTopicChanged, 1)
	}, nil)
}
watch topics if changed
package consistence
import (
	"encoding/json"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/coreos/etcd/client"
	lru "github.com/hashicorp/golang-lru"
	"golang.org/x/net/context"
)
const (
	EVENT_WATCH_TOPIC_L_CREATE = iota
	EVENT_WATCH_TOPIC_L_DELETE
)
type cachedItemWithTs struct {
	value interface{}
	ts    int64
}
func getCacheValue(cache *lru.ARCCache, k string, tn time.Time) (interface{}, bool) {
	cv, ok := cache.Get(k)
	if ok {
		item := cv.(cachedItemWithTs)
		if item.ts > tn.UnixNano() {
			return item.value, true
		}
	}
	return nil, false
}
func putCacheValue(cache *lru.ARCCache, k string, v interface{}, tn time.Time, ttl time.Duration) {
	cache.Add(k, cachedItemWithTs{
		ts:    tn.Add(ttl).UnixNano(),
		value: v,
	})
}
type NsqLookupdEtcdMgr struct {
	tmiMutex sync.RWMutex
	cache    *lru.ARCCache
	client              *EtcdClient
	clusterID           string
	topicRoot           string
	clusterPath         string
	leaderSessionPath   string
	leaderStr           string
	lookupdRootPath     string
	topicMetaInfos      []TopicPartitionMetaInfo
	currentClusterIndex uint64
	topicMetaMap        map[string]TopicMetaInfo
	ifTopicChanged      int32
	ifTopicScanning     int32
	nodeInfo            *NsqLookupdNodeInfo
	nodeKey             string
	nodeValue           string
	refreshStopCh        chan bool
	watchTopicsStopCh    chan struct{}
	watchNsqdNodesStopCh chan bool
	topicReplicasMap    map[string]map[int]TopicPartitionReplicaInfo
	watchedClusterIndex uint64
}
func NewNsqLookupdEtcdMgr(host, username, pwd string) (*NsqLookupdEtcdMgr, error) {
	client, err := NewEClient(host, username, pwd)
	if err != nil {
		return nil, err
	}
	c, _ := lru.NewARC(10000)
	return &NsqLookupdEtcdMgr{
		client:               client,
		ifTopicChanged:       1,
		ifTopicScanning:      0,
		watchTopicsStopCh:    make(chan struct{}, 1),
		watchNsqdNodesStopCh: make(chan bool, 1),
		topicMetaMap:         make(map[string]TopicMetaInfo),
		refreshStopCh:        make(chan bool, 1),
		topicReplicasMap:     make(map[string]map[int]TopicPartitionReplicaInfo),
		cache:                c,
	}, nil
}
func (self *NsqLookupdEtcdMgr) InitClusterID(id string) {
	self.clusterID = id
	self.topicRoot = self.createTopicRootPath()
	self.clusterPath = self.createClusterPath()
	self.leaderSessionPath = self.createLookupdLeaderPath()
	self.lookupdRootPath = self.createLookupdRootPath()
	go self.watchTopics()
}
func (self *NsqLookupdEtcdMgr) Register(value *NsqLookupdNodeInfo) error {
	self.nodeInfo = value
	valueB, err := json.Marshal(value)
	if err != nil {
		return err
	}
	if self.refreshStopCh != nil {
		close(self.refreshStopCh)
	}
	self.leaderStr = string(valueB)
	self.nodeKey = self.createLookupdPath(value)
	self.nodeValue = string(valueB)
	_, err = self.client.Set(self.nodeKey, self.nodeValue, ETCD_TTL)
	if err != nil {
		return err
	}
	self.refreshStopCh = make(chan bool, 1)
	go self.refresh(self.refreshStopCh)
	return nil
}
func (self *NsqLookupdEtcdMgr) refresh(stopC <-chan bool) {
	ticker := time.NewTicker(time.Second * time.Duration(ETCD_TTL/10))
	defer ticker.Stop()
	for {
		select {
		case <-stopC:
			return
		case <-ticker.C:
			_, err := self.client.SetWithTTL(self.nodeKey, ETCD_TTL)
			if err != nil {
				coordLog.Errorf("update error: %s", err.Error())
				_, err := self.client.Set(self.nodeKey, self.nodeValue, ETCD_TTL)
				if err != nil {
					coordLog.Errorf("set key error: %s", err.Error())
				}
			}
		}
	}
}
func (self *NsqLookupdEtcdMgr) Unregister(value *NsqLookupdNodeInfo) error {
	if self.refreshStopCh != nil {
		close(self.refreshStopCh)
		self.refreshStopCh = nil
	}
	_, err := self.client.Delete(self.createLookupdPath(value), false)
	if err != nil {
		coordLog.Warningf("cluser[%v] node[%v] unregister failed: %v", self.clusterID, value, err)
		return err
	}
	return nil
}
func (self *NsqLookupdEtcdMgr) Stop() {
	coordLog.Infof("lookup etcd leadership is stopping.")
	if self.watchNsqdNodesStopCh != nil {
		close(self.watchNsqdNodesStopCh)
	}
	if self.watchTopicsStopCh != nil {
		close(self.watchTopicsStopCh)
	}
}
func (self *NsqLookupdEtcdMgr) GetClusterEpoch() (EpochType, error) {
	rsp, err := self.client.Get(self.clusterPath, false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return 0, ErrKeyNotFound
		}
		return 0, err
	}
	return EpochType(rsp.Node.ModifiedIndex), nil
}
func (self *NsqLookupdEtcdMgr) GetAllLookupdNodes() ([]NsqLookupdNodeInfo, error) {
	tn := time.Now()
	cv, ok := getCacheValue(self.cache, self.lookupdRootPath, tn)
	if ok {
		return cv.([]NsqLookupdNodeInfo), nil
	}
	rsp, err := self.client.Get(self.lookupdRootPath, false, false)
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	lookupdNodeList := make([]NsqLookupdNodeInfo, 0)
	for _, node := range rsp.Node.Nodes {
		var nodeInfo NsqLookupdNodeInfo
		if err = json.Unmarshal([]byte(node.Value), &nodeInfo); err != nil {
			continue
		}
		lookupdNodeList = append(lookupdNodeList, nodeInfo)
	}
	putCacheValue(self.cache, self.lookupdRootPath, lookupdNodeList, tn, time.Second)
	return lookupdNodeList, nil
}
func (self *NsqLookupdEtcdMgr) AcquireAndWatchLeader(leader chan *NsqLookupdNodeInfo, stop chan struct{}) {
	master := NewMaster(self.client, self.leaderSessionPath, self.leaderStr, ETCD_TTL)
	go self.processMasterEvents(master, leader, stop)
	master.Start()
}
func (self *NsqLookupdEtcdMgr) GetTopicsMetaInfoMap(topics []string) (map[string]TopicMetaInfo, error) {
	topicMetaInfoCache := make(map[string]TopicMetaInfo)
	if atomic.LoadInt32(&self.ifTopicChanged) == 1 {
		for _, topic := range topics {
			topicMeta, _, err := self.GetTopicMetaInfo(topic)
			if err != nil {
				return nil, err
			}
			topicMetaInfoCache[topic] = topicMeta
		}
	} else {
		self.tmiMutex.RLock()
		defer self.tmiMutex.RUnlock()
		for _, topic := range topics {
			topicMeta, exist := self.topicMetaMap[topic]
			if !exist {
				topicMetaInfoCache[topic] = TopicMetaInfo{}
			} else {
				topicMetaInfoCache[topic] = topicMeta
			}
		}
	}
	return topicMetaInfoCache, nil
}
func (self *NsqLookupdEtcdMgr) processMasterEvents(master Master, leader chan *NsqLookupdNodeInfo, stop chan struct{}) {
	for {
		select {
		case e := <-master.GetEventsChan():
			if e.Type == MASTER_ADD || e.Type == MASTER_MODIFY {
				var lookupdNode NsqLookupdNodeInfo
				if err := json.Unmarshal([]byte(e.Master), &lookupdNode); err != nil {
					leader <- &lookupdNode
					continue
				}
				coordLog.Infof("master event type[%d] lookupdNode[%v].", e.Type, lookupdNode)
				leader <- &lookupdNode
			} else if e.Type == MASTER_DELETE {
				coordLog.Infof("master event delete.")
				var lookupdNode NsqLookupdNodeInfo
				leader <- &lookupdNode
			} else {
				coordLog.Infof("master event unknown: %v.", e)
			}
		case <-stop:
			master.Stop()
			close(leader)
			return
		}
	}
}
func (self *NsqLookupdEtcdMgr) GetNsqdNodes() ([]NsqdNodeInfo, error) {
	nodes, _, err := self.getNsqdNodes(false)
	return nodes, err
}
func (self *NsqLookupdEtcdMgr) WatchNsqdNodes(nsqds chan []NsqdNodeInfo, stop chan struct{}) {
	defer close(nsqds)
	nsqdNodes, _, err := self.getNsqdNodes(false)
	if err == nil {
		select {
		case nsqds <- nsqdNodes:
		case <-stop:
			return
		}
	}
	key := self.createNsqdRootPath()
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		select {
		case <-stop:
			coordLog.Infof("watch key[%s] will be stopped by 1.", key)
			cancel()
		case <-self.watchNsqdNodesStopCh:
			coordLog.Infof("watch key[%s] will be stopped by 2.", key)
			cancel()
		}
	}()
	watchWaitAndDo(ctx, self.client, key, true, func(rsp *client.Response) {
		nsqdNodes, _, err := self.getNsqdNodes(true)
		if err != nil {
			coordLog.Errorf("key[%s] getNsqdNodes error: %s", key, err.Error())
			return
		}
		select {
		case nsqds <- nsqdNodes:
		case <-stop:
			return
		}
	}, nil)
}
func (self *NsqLookupdEtcdMgr) getNsqdNodes(upToDate bool) ([]NsqdNodeInfo, uint64, error) {
	var rsp *client.Response
	var err error
	if upToDate {
		rsp, err = self.client.GetNewest(self.createNsqdRootPath(), false, false)
	} else {
		rsp, err = self.client.Get(self.createNsqdRootPath(), false, false)
	}
	if err != nil {
		if client.IsKeyNotFound(err) {
			return nil, 0, ErrKeyNotFound
		}
		return nil, 0, err
	}
	nsqdNodes := make([]NsqdNodeInfo, 0)
	for _, node := range rsp.Node.Nodes {
		if node.Dir {
			continue
		}
		var nodeInfo NsqdNodeInfo
		err := json.Unmarshal([]byte(node.Value), &nodeInfo)
		if err != nil {
			continue
		}
		nsqdNodes = append(nsqdNodes, nodeInfo)
	}
	return nsqdNodes, rsp.Index, err
}
func (self *NsqLookupdEtcdMgr) GetAllTopicMetas() (map[string]TopicMetaInfo, error) {
	self.tmiMutex.RLock()
	topicMetas := self.topicMetaMap
	self.tmiMutex.RUnlock()
	return topicMetas, nil
}
func (self *NsqLookupdEtcdMgr) ScanTopics() ([]TopicPartitionMetaInfo, error) {
	if atomic.LoadInt32(&self.ifTopicChanged) == 1 {
		return self.scanTopics()
	}
	self.tmiMutex.RLock()
	topicMetaInfos := self.topicMetaInfos
	self.tmiMutex.RUnlock()
	return topicMetaInfos, nil
}
func getMaxIndexFromNode(n *client.Node) uint64 {
	if n == nil {
		return 0
	}
	maxI := n.ModifiedIndex
	if n.Dir {
		for _, child := range n.Nodes {
			index := getMaxIndexFromNode(child)
			if index > maxI {
				maxI = index
			}
		}
	}
	return maxI
}
func getMaxIndexFromWatchRsp(resp *client.Response) uint64 {
	if resp == nil {
		return 0
	}
	index := resp.Index
	if resp.Node != nil {
		if nmi := getMaxIndexFromNode(resp.Node); nmi > index {
			index = nmi
		}
	}
	return index
}
func watchWaitAndDo(ctx context.Context, client *EtcdClient,
	key string, recursive bool, callback func(rsp *client.Response),
	watchExpiredCb func()) {
	initIndex := uint64(0)
	rsp, err := client.Get(key, false, recursive)
	if err != nil {
		coordLog.Errorf("get watched key[%s] error: %s", key, err.Error())
	} else {
		if rsp.Index > 0 {
			initIndex = rsp.Index - 1
		}
		callback(rsp)
	}
	watcher := client.Watch(key, initIndex, recursive)
	for {
		rsp, err := watcher.Next(ctx)
		if err != nil {
			if err == context.Canceled {
				coordLog.Infof("watch key[%s] cancelled.", key)
				return
			} else if err == context.DeadlineExceeded {
				coordLog.Debugf("watcher key[%s] timeout: %s", key, err.Error())
				continue
			} else {
				if IsEtcdWatchExpired(err) {
					coordLog.Debugf("watcher key[%s] error: %s", key, err.Error())
					if watchExpiredCb != nil {
						watchExpiredCb()
					}
					rsp, err = client.Get(key, false, false)
					if err != nil {
						coordLog.Errorf("rewatch and get key[%s] error: %s", key, err.Error())
						time.Sleep(time.Second)
						continue
					}
					watcher = client.Watch(key, rsp.Index, recursive)
				} else {
					coordLog.Errorf("watcher key[%s] error: %s", key, err.Error())
					time.Sleep(5 * time.Second)
					continue
				}
			}
		}
		callback(rsp)
	}
}