func (k *KVStore) listenForLeader(name string, callback func(name, uri string)) {
	pair, meta, err := k.kv.Get(fmt.Sprintf("%s-%s", Prefix, name), nil)
	if err != nil {
		k.log.Panic("Unable to list keys", err)
	}

	if pair != nil {
		callback(stripLeaderPrefix(pair.Key), string(pair.Value))
	}

	options := api.QueryOptions{
		RequireConsistent: true,
	}
	for {
		options.WaitIndex = meta.LastIndex
		pair, meta, err = k.kv.Get(fmt.Sprintf("%s-%s", Prefix, name), &options)
		if err != nil {
			k.log.Error("Unable to get leader", err)
			return
		}

		if pair != nil {
			callback(stripLeaderPrefix(pair.Key), string(pair.Value))
		}
	}
}

func (data *consulData) getServiceEntries(health *api.Health) (entries []*api.ServiceEntry, err error) {
	options := api.QueryOptions{}
	if data.dc != "" {
		options.Datacenter = data.dc
	}

	entries, _, err = health.Service(data.serviceName, data.serviceTag, true, &options)
	return entries, err
}

func (m *Meta) QueryOptions() *consulapi.QueryOptions {
	queryOpts := new(consulapi.QueryOptions)
	if m.token != "" {
		queryOpts.Token = m.token
	}

	if m.dc != "" {
		queryOpts.Datacenter = m.dc
	}

	if m.waitIndex != 0 {
		queryOpts.WaitIndex = m.waitIndex
	}

	return queryOpts
}

// GetWatcher watches for kvstore changes in the given key. Triggers the returned channel
// every time the key path is changed.
func (c *ConsulClient) GetWatcher(key string, timeSleep time.Duration) <-chan []uint32 {
	ch := make(chan []uint32, 100)
	go func() {
		curSeconds := time.Second
		var (
			k   *consulAPI.KVPair
			q   *consulAPI.QueryMeta
			qo  consulAPI.QueryOptions
			err error
		)
		for {
			k, q, err = c.KV().Get(key, nil)
			if err != nil {
				log.Errorf("Unable to retrieve last free Index: %s", err)
			}
			if k != nil {
				break
			} else {
				log.Debugf("Unable to retrieve last free Index, please start some containers with labels.")
			}
			time.Sleep(timeSleep)
		}

		for {
			k, q, err = c.KV().Get(key, &qo)
			if err != nil {
				log.Errorf("Unable to retrieve last free Index: %s", err)
			}
			if k == nil || q == nil {
				log.Warning("Unable to retrieve last free Index, please start some containers with labels.")
				time.Sleep(curSeconds)
				if curSeconds < timeSleep {
					curSeconds += time.Second
				}
				continue
			}
			curSeconds = time.Second
			qo.WaitIndex = q.LastIndex
			go func() {
				ch <- []uint32{}
			}()
		}
	}()
	return ch
}

func (m *Meta) QueryOptions() *consulapi.QueryOptions {
	queryOpts := new(consulapi.QueryOptions)

	if os.Getenv("CONSUL_TOKEN") != "" {
		queryOpts.Token = os.Getenv("CONSUL_TOKEN")
	}

	if m.token != "" {
		queryOpts.Token = m.token
	}

	if m.dc != "" {
		queryOpts.Datacenter = m.dc
	}

	if m.waitIndex != 0 {
		queryOpts.WaitIndex = m.waitIndex
	}

	return queryOpts
}

func (c *Cmd) QueryOptions() *consulapi.QueryOptions {
	csl := c.consul

	queryOpts := new(consulapi.QueryOptions)
	if csl.token != "" {
		queryOpts.Token = csl.token
	}

	if csl.dc != "" {
		queryOpts.Datacenter = csl.dc
	}

	if csl.waitIndex != 0 {
		queryOpts.WaitIndex = csl.waitIndex
	}

	if csl.consistent {
		queryOpts.RequireConsistent = csl.consistent
	}

	if csl.stale {
		queryOpts.AllowStale = csl.stale
	}

	return queryOpts
}

func (k *KVStore) listenForAnnouncements(callback func(name string)) {
	pairs, meta, err := k.kv.List(AnnouncePrefix, nil)
	if err != nil {
		k.log.Panic("Unable to list keys", err)
	}

	for _, pair := range pairs {
		callback(string(pair.Value))
	}

	var options api.QueryOptions
	for {
		options.WaitIndex = meta.LastIndex
		pairs, meta, err = k.kv.List(AnnouncePrefix, &options)
		if err != nil {
			k.log.Error("Unable to list keys", err)
			return
		}

		for _, pair := range pairs {
			callback(string(pair.Value))
		}
	}
}

func consulQuery(service string, tag string, client *api.Client, options *api.QueryOptions, channel chan []*api.CatalogService) {
	catalog := client.Catalog()
	failures := 0
	for {
		nodes, qm, err := catalog.Service(service, tag, options)
		if err != nil {
			failures++
			retry := retryInterval * time.Duration(failures*failures)
			if retry > maxBackoffTime {
				retry = maxBackoffTime
			}
			log.Printf("Consul monitor errored: %s, retry in %s", err, retry)
			<-time.After(retry)
			continue
		}
		failures = 0
		if options.WaitIndex == qm.LastIndex {
			continue
		}
		options.WaitIndex = qm.LastIndex
		channel <- nodes
	}
}

// streamResults is used to perform blocking queries against the KV endpoint and stream in
// notice of various events into waitForJob
func (c *ExecCommand) streamResults(doneCh chan struct{}, ackCh chan rExecAck, heartCh chan rExecHeart,
	outputCh chan rExecOutput, exitCh chan rExecExit, errCh chan struct{}) {
	kv := c.client.KV()
	opts := consulapi.QueryOptions{WaitTime: c.conf.wait}
	dir := path.Join(c.conf.prefix, c.sessionID) + "/"
	seen := make(map[string]struct{})

	for {
		// Check if we've been signaled to exit
		select {
		case <-doneCh:
			return
		default:
		}

		// Block on waiting for new keys
		keys, qm, err := kv.Keys(dir, "", &opts)
		if err != nil {
			c.Ui.Error(fmt.Sprintf("Failed to read results: %s", err))
			goto ERR_EXIT
		}

		// Fast-path the no-change case
		if qm.LastIndex == opts.WaitIndex {
			continue
		}
		opts.WaitIndex = qm.LastIndex

		// Handle each key
		for _, key := range keys {
			// Ignore if we've seen it
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}

			// Trim the directory
			full := key
			key = strings.TrimPrefix(key, dir)

			// Handle the key type
			switch {
			case key == rExecFileName:
				continue
			case strings.HasSuffix(key, rExecAckSuffix):
				ackCh <- rExecAck{Node: strings.TrimSuffix(key, rExecAckSuffix)}

			case strings.HasSuffix(key, rExecExitSuffix):
				pair, _, err := kv.Get(full, nil)
				if err != nil || pair == nil {
					c.Ui.Error(fmt.Sprintf("Failed to read key '%s': %v", full, err))
					continue
				}
				code, err := strconv.ParseInt(string(pair.Value), 10, 32)
				if err != nil {
					c.Ui.Error(fmt.Sprintf("Failed to parse exit code '%s': %v", pair.Value, err))
					continue
				}
				exitCh <- rExecExit{
					Node: strings.TrimSuffix(key, rExecExitSuffix),
					Code: int(code),
				}

			case strings.LastIndex(key, rExecOutputDivider) != -1:
				pair, _, err := kv.Get(full, nil)
				if err != nil || pair == nil {
					c.Ui.Error(fmt.Sprintf("Failed to read key '%s': %v", full, err))
					continue
				}
				idx := strings.LastIndex(key, rExecOutputDivider)
				node := key[:idx]
				if len(pair.Value) == 0 {
					heartCh <- rExecHeart{Node: node}
				} else {
					outputCh <- rExecOutput{Node: node, Output: pair.Value}
				}

			default:
				c.Ui.Error(fmt.Sprintf("Unknown key '%s', ignoring.", key))
			}
		}
	}

ERR_EXIT:
	select {
	case errCh <- struct{}{}:
	default:
	}
}

func (c *ConsulRegistry) WatchConsul() {
	c.Map = make(map[string][]string, 0)
	c.Matcher = make(map[string]*MyRegExp, 0)

	client, err := api.NewClient(api.DefaultConfig())
	if err != nil {
		log.Fatal("Failed to attach to consul agent: ", err)
	}

	kv := client.KV()
	catalog := client.Catalog()
	q := api.QueryOptions{
		WaitIndex: 0,
		WaitTime:  time.Second * 10,
	}

	knownServices := make(map[string]*api.CatalogService)
	for {
		services, meta, err := catalog.Services(&q)
		if err != nil {
			log.Fatal("Failed to get service catalog from consul agent: ", err)
		}
		wantedServices := make(map[string]*api.CatalogService)
		toRemoveServices := make(map[string]*api.CatalogService)
		untouchedServices := make(map[string]*api.CatalogService)
		for svcName := range services {
			svcCatalog, _, err := catalog.Service(svcName, "", nil)
			if err != nil {
				log.Fatal("Failed to get service entry from consul agent: ", err)
			}
			if len(svcCatalog) == 0 {
				continue
			}
			svc := svcCatalog[0]
			if !stringInSlice("revproxy", svc.ServiceTags) {
				continue
			}
			if forwarderMode && strings.HasPrefix(svcName, "internal-") {
				continue
			}
			if stringInSlice("revproxy-default", svc.ServiceTags) {
				c.Default = svcName
			}

			// Bucketize the services we want to forward
			if knownSvc, ok := knownServices[svcName]; ok {
				if knownSvc.Address == svc.Address &&
					knownSvc.ServiceAddress == svc.ServiceAddress &&
					knownSvc.ServicePort == svc.ServicePort {
					// Nothing changed, it goes in the untouched bucket.
					untouchedServices[svcName] = knownSvc
				} else {
					// Something changed, it goes in the toRemove and
					// wanted buckets
					log.Printf("%s has changed config, will update forwarding rules", svcName)
					log.Printf("%s: ServiceAddress %v => %v", svcName, knownSvc.ServiceAddress, svc.ServiceAddress)
					log.Printf("%s: ServicePort %v => %v", svcName, knownSvc.ServicePort, svc.ServicePort)
					log.Printf("%s: Address %v => %v", svcName, knownSvc.Address, svc.Address)
					toRemoveServices[svcName] = knownSvc
					wantedServices[svcName] = svc
				}
			} else {
				log.Printf("%s is new, will add to rules", svcName)
				log.Printf("%s: ServiceAddress %v", svcName, svc.ServiceAddress)
				log.Printf("%s: ServicePort %v", svcName, svc.ServicePort)
				log.Printf("%s: Address %v", svcName, svc.Address)
				// New service, it goes in the wanted bucket
				wantedServices[svcName] = svc
			}
		}
		// Any known services that are not in the wanted or untouched bucket
		// need to be removed.
		for svcName, svc := range knownServices {
			if _, ok := wantedServices[svcName]; ok {
				continue
			}
			if _, ok := untouchedServices[svcName]; ok {
				continue
			}
			log.Printf("%s has gone away, will delete forwarding rules", svcName)
			toRemoveServices[svcName] = svc
		}
		// Delete services we no longer care about
		for svcTag, svc := range toRemoveServices {
			// Whack service registration iff the service was removed.
			svcAddr := svc.ServiceAddress
			if svcAddr == "" {
				svcAddr = svc.Address
			}
			svcPort := fmt.Sprintf("%d", svc.ServicePort)

			c.Delete(svcTag, svcAddr+":"+svcPort)
		}
		// Add new services we do care about
		for svcTag, svc := range wantedServices {
			svcAddr := svc.ServiceAddress
			if svcAddr == "" {
				svcAddr = svc.Address
			}
			svcMatcher := "jjk"
//.........这里部分代码省略.........