// WatchAllState watches all state from the baseKey.
966484
func (d *EtcdStateDriver) WatchAllState(baseKey string, sType core.State,
	unmarshal func([]byte, interface{}) error, rsps chan core.WatchState) error {
	byteRsps := make(chan [2][]byte, 1)
	recvErr := make(chan error, 1)

	err := d.WatchAll(baseKey, byteRsps)
	if err != nil {
		log.Errorf("WatchAll returned %v", err)
		return err
	}

	for {
		go channelStateEvents(d, sType, unmarshal, byteRsps, rsps, recvErr)

		err = <-recvErr
		log.Errorf("Err from channelStateEvents %v", err)
		time.Sleep(time.Second)
	}
}
188559
func (d *Downloader) syncState(root common.Hash) *stateSync {
	s := newStateSync(d, root)
	select {
	case d.stateSyncStart <- s:
	case <-d.quitCh:
		s.err = errCancelStateFetch
		close(s.done)
	}
	return s
}
54635
func (s *EndpointGroupState) WatchAll(rsps chan core.WatchState) error {
	return s.StateDriver.WatchAllState(epGroupConfigPathPrefix, s, json.Unmarshal,
		rsps)
}
56323
func (st *State) WatchControllerStatusChanges() StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    statusesC,
		filter: makeControllerIdFilter(st),
	})
}
1255936
func (s *StateStore) Allocs(ws memdb.WatchSet) (memdb.ResultIterator, error) {
	txn := s.db.Txn(false)

	// Walk the entire table
	iter, err := txn.Get("allocs", "id")
	if err != nil {
		return nil, err
	}

	ws.Add(iter.WatchCh())

	return iter, nil
}
224546
func (g *Oper) ReadAll() ([]core.State, error) {
	return g.StateDriver.ReadAllState(operGlobalPrefix, g, json.Unmarshal)
}
167419
func (d *FakeStateDriver) ReadAll(baseKey string) ([][]byte, error) {
	values := [][]byte{}

	for key, val := range d.TestState {
		if strings.Contains(key, baseKey) {
			values = append(values, val.value)
		}
	}
	return values, nil
}
564225
func (r *Relation) WatchUnits(appName string) (RelationUnitsWatcher, error) {
	return r.watchUnits(appName, false)
}
229226
func (c *consulClient) Watch(w *Watcher) {
	// Last known state of all KVPairs matching the prefix
	localState := map[string]consulAPI.KVPair{}
	nextIndex := uint64(0)

	qo := consulAPI.QueryOptions{
		WaitTime: time.Second,
	}

	for {
		// Initialize sleep time to a millisecond as we don't
		// want to sleep in between successful watch cycles
		sleepTime := 1 * time.Millisecond

		qo.WaitIndex = nextIndex
		pairs, q, err := c.KV().List(w.prefix, &qo)
		if err != nil {
			sleepTime = 5 * time.Second
			Trace("List of Watch failed", err, logrus.Fields{fieldPrefix: w.prefix, fieldWatcher: w.name})
		}

		if q != nil {
			nextIndex = q.LastIndex
		}

		// timeout while watching for changes, re-schedule
		if qo.WaitIndex != 0 && (q == nil || q.LastIndex == qo.WaitIndex) {
			goto wait
		}

		for _, newPair := range pairs {
			oldPair, ok := localState[newPair.Key]

			// Keys reported for the first time must be new
			if !ok {
				if newPair.CreateIndex != newPair.ModifyIndex {
					log.Debugf("consul: Previously unknown key %s received with CreateIndex(%d) != ModifyIndex(%d)",
						newPair.Key, newPair.CreateIndex, newPair.ModifyIndex)
				}

				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Typ:   EventTypeCreate,
					Key:   newPair.Key,
					Value: newPair.Value,
				}
				trackEventQueued(newPair.Key, EventTypeCreate, queueStart.End(true).Total())
			} else if oldPair.ModifyIndex != newPair.ModifyIndex {
				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Typ:   EventTypeModify,
					Key:   newPair.Key,
					Value: newPair.Value,
				}
				trackEventQueued(newPair.Key, EventTypeModify, queueStart.End(true).Total())
			}

			// Everything left on localState will be assumed to
			// have been deleted, therefore remove all keys in
			// localState that still exist in the kvstore
			delete(localState, newPair.Key)
		}

		for k, deletedPair := range localState {
			queueStart := spanstat.Start()
			w.Events <- KeyValueEvent{
				Typ:   EventTypeDelete,
				Key:   deletedPair.Key,
				Value: deletedPair.Value,
			}
			trackEventQueued(deletedPair.Key, EventTypeDelete, queueStart.End(true).Total())
			delete(localState, k)
		}

		for _, newPair := range pairs {
			localState[newPair.Key] = *newPair

		}

		// Initial list operation has been completed, signal this
		if qo.WaitIndex == 0 {
			w.Events <- KeyValueEvent{Typ: EventTypeListDone}
		}

	wait:
		select {
		case <-time.After(sleepTime):
		case <-w.stopWatch:
			close(w.Events)
			w.stopWait.Done()
			return
		}
	}
}
421630
func (s *Etcd) Watch(key string, ctx context.Context, recursive bool, index uint64) (<-chan *store.Event, error) {
	opts := &etcd.WatcherOptions{Recursive: recursive, AfterIndex: index}
	watcher := s.client.Watcher(s.normalize(key), opts)

	// watchCh is sending back events to the caller
	watchCh := make(chan *store.Event)

	go func() {
		defer close(watchCh)
		for {
			result, err := watcher.Next(ctx)

			if err != nil {
				// Push error value through the channel.
				watchCh <- errorEvent(key, err)
				return
			}
			watchCh <- &store.Event{
				Action:        actionMap[result.Action],
				Key:           result.Node.Key,
				ModifiedIndex: result.Node.ModifiedIndex,
				Data: []*store.KVPair{
					&store.KVPair{
						Key:       result.Node.Key,
						Value:     []byte(result.Node.Value),
						LastIndex: result.Node.ModifiedIndex,
					},
				},
			}
		}
	}()

	return watchCh, nil
}
1366359
func (st *State) WatchSubnets(subnetFilter func(id interface{}) bool) StringsWatcher {
	filter := func(id interface{}) bool {
		subnet, err := st.strictLocalID(id.(string))
		if err != nil {
			return false
		}
		if subnetFilter == nil {
			return true
		}
		return subnetFilter(subnet)
	}

	return newLifecycleWatcher(st, subnetsC, nil, filter, nil)
}
710944
func (c *Client) WatchUnits(application string) (watcher.StringsWatcher, error) {
	applicationTag, err := applicationTag(application)
	if err != nil {
		return nil, errors.Trace(err)
	}
	args := entities(applicationTag)

	var results params.StringsWatchResults
	if err := c.facade.FacadeCall("WatchUnits", args, &results); err != nil {
		return nil, err
	}
	if n := len(results.Results); n != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", n)
	}
	if err := results.Results[0].Error; err != nil {
		return nil, errors.Trace(err)
	}
	w := apiwatcher.NewStringsWatcher(c.facade.RawAPICaller(), results.Results[0])
	return w, nil
}
933179
func (s *state) initWatches() error {
	// Watch for root changes
	err := s.cache.Notify(s.ctx, cachetype.ConnectCARootName, &structs.DCSpecificRequest{
		Datacenter:   s.source.Datacenter,
		QueryOptions: structs.QueryOptions{Token: s.token},
	}, rootsWatchID, s.ch)
	if err != nil {
		return err
	}

	// Watch the leaf cert
	err = s.cache.Notify(s.ctx, cachetype.ConnectCALeafName, &cachetype.ConnectCALeafRequest{
		Datacenter: s.source.Datacenter,
		Token:      s.token,
		Service:    s.proxyCfg.DestinationServiceName,
	}, leafWatchID, s.ch)
	if err != nil {
		return err
	}

	// Watch for intention updates
	err = s.cache.Notify(s.ctx, cachetype.IntentionMatchName, &structs.IntentionQueryRequest{
		Datacenter:   s.source.Datacenter,
		QueryOptions: structs.QueryOptions{Token: s.token},
		Match: &structs.IntentionQueryMatch{
			Type: structs.IntentionMatchDestination,
			Entries: []structs.IntentionMatchEntry{
				{
					Namespace: structs.IntentionDefaultNamespace,
					Name:      s.proxyCfg.DestinationServiceName,
				},
			},
		},
	}, intentionsWatchID, s.ch)
	if err != nil {
		return err
	}

	// Watch for updates to service endpoints for all upstreams
	for _, u := range s.proxyCfg.Upstreams {
		dc := s.source.Datacenter
		if u.Datacenter != "" {
			dc = u.Datacenter
		}

		switch u.DestinationType {
		case structs.UpstreamDestTypePreparedQuery:
			err = s.cache.Notify(s.ctx, cachetype.PreparedQueryName, &structs.PreparedQueryExecuteRequest{
				Datacenter:    dc,
				QueryOptions:  structs.QueryOptions{Token: s.token, MaxAge: defaultPreparedQueryPollInterval},
				QueryIDOrName: u.DestinationName,
				Connect:       true,
			}, "upstream:"+u.Identifier(), s.ch)
		case structs.UpstreamDestTypeService:
			fallthrough
		case "": // Treat unset as the default Service type
			err = s.cache.Notify(s.ctx, cachetype.HealthServicesName, &structs.ServiceSpecificRequest{
				Datacenter:   dc,
				QueryOptions: structs.QueryOptions{Token: s.token},
				ServiceName:  u.DestinationName,
				Connect:      true,
				// Note that Identifier doesn't type-prefix for service any more as it's
				// the default and makes metrics and other things much cleaner. It's
				// simpler for us if we have the type to make things unambiguous.
			}, "upstream:"+serviceIDPrefix+u.Identifier(), s.ch)

			if err != nil {
				return err
			}

		default:
			return fmt.Errorf("unknown upstream type: %q", u.DestinationType)
		}
	}
	return nil
}
81787
func (c *cache) watch(w registry.Watcher) error {
	defer w.Stop()

	// manage this loop
	go func() {
		// wait for exit
		<-c.exit
		w.Stop()
	}()

	for {
		res, err := w.Next()
		if err != nil {
			return err
		}
		c.update(res)
	}
}
930647
func (s *CfgServiceLBState) ReadAll() ([]core.State, error) {
	return s.StateDriver.ReadAllState(serviceLBConfigPathPrefix, s, json.Unmarshal)
}
896319
func (s *GlobConfig) WatchAll(rsps chan core.WatchState) error {
	return s.StateDriver.WatchAllState(globalConfigPathPrefix, s, json.Unmarshal,
		rsps)
}
1177876
func NewStateAlloc(f Alloc) *State {
	ls := C.clua_newstate(unsafe.Pointer(&f))
	return newState(ls)
}