// WatchAllState reads all state from baseKey of a given type
343676
func (d *FakeStateDriver) WatchAllState(baseKey string, sType core.State,
	unmarshal func([]byte, interface{}) error, rsps chan core.WatchState) error {
	return core.Errorf("not supported")
}
564225
func (r *Relation) WatchUnits(appName string) (RelationUnitsWatcher, error) {
	return r.watchUnits(appName, false)
}
46307
func (s *SvcProvider) ReadAll() ([]core.State, error) {
	return s.StateDriver.ReadAllState(svcProviderPathPrefix, s, json.Unmarshal)
}
1373971
func (b *Driver) GenerateAllStates() (err error) {
	b.stepsfa0b++
	buf := []byte{0x04, byte(b.stepsfa0b), 0x00, 0x04, 0x01, 0x00, 0x32, 0x30, 0x31, 0x34, 0x2D, 0x31, 0x30, 0x2D, 0x32, 0x38, 0x00}
	err = b.adaptor().WriteCharacteristic(commandCharacteristic, buf)

	return
}
665306
func (s *CfgEndpointState) ReadAll() ([]core.State, error) {
	return s.StateDriver.ReadAllState(endpointConfigPathPrefix, s, json.Unmarshal)
}
204784
func (s *EndpointGroupState) ReadAll() ([]core.State, error) {
	return s.StateDriver.ReadAllState(epGroupConfigPathPrefix, s, json.Unmarshal)
}
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
		}
	}
}
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
}
1333658
func (r *Router) watchHistory() {
	js.Global.Set("onpopstate", func() {
		go func() {
			r.pathChanged(getPath(), false)
			if r.ShouldInterceptLinks {
				r.InterceptLinks()
			}
		}()
	})
}
323637
func (p *ProvisionerAPI) WatchAllContainers(args params.WatchContainers) (params.StringsWatchResults, error) {
	return p.WatchContainers(args)
}
919939
func readAllStateCommon(d core.StateDriver, baseKey string, sType core.State,
	unmarshal func([]byte, interface{}) error) ([]core.State, error) {
	stateType := reflect.TypeOf(sType)
	sliceType := reflect.SliceOf(stateType)
	values := reflect.MakeSlice(sliceType, 0, 1)

	byteValues, err := d.ReadAll(baseKey)
	if err != nil {
		return nil, err
	}
	for _, byteValue := range byteValues {
		value := reflect.New(stateType)
		err = unmarshal(byteValue, value.Interface())
		if err != nil {
			return nil, err
		}
		values = reflect.Append(values, value.Elem())
	}

	stateValues := []core.State{}
	for i := 0; i < values.Len(); i++ {
		// sanity checks
		if !values.Index(i).Elem().FieldByName("CommonState").IsValid() {
			return nil, core.Errorf("The state structure %v is missing core.CommonState",
				stateType)
		}
		//the following works as every core.State is expected to embed core.CommonState struct
		values.Index(i).Elem().FieldByName("CommonState").FieldByName("StateDriver").Set(reflect.ValueOf(d))
		stateValue := values.Index(i).Interface().(core.State)
		stateValues = append(stateValues, stateValue)
	}
	return stateValues, nil
}
1157070
func makeAllWatcherCollectionInfo(collNames ...string) map[string]allWatcherStateCollection {
	seenTypes := make(map[reflect.Type]struct{})
	collectionByName := make(map[string]allWatcherStateCollection)

	for _, collName := range collNames {
		collection := allWatcherStateCollection{name: collName}
		switch collName {
		case modelsC:
			collection.docType = reflect.TypeOf(backingModel{})
		case machinesC:
			collection.docType = reflect.TypeOf(backingMachine{})
		case instanceDataC:
			collection.docType = reflect.TypeOf(backingInstanceData{})
			collection.subsidiary = true
		case unitsC:
			collection.docType = reflect.TypeOf(backingUnit{})
		case applicationsC:
			collection.docType = reflect.TypeOf(backingApplication{})
		case charmsC:
			collection.docType = reflect.TypeOf(backingCharm{})
		case actionsC:
			collection.docType = reflect.TypeOf(backingAction{})
		case relationsC:
			collection.docType = reflect.TypeOf(backingRelation{})
		case annotationsC:
			collection.docType = reflect.TypeOf(backingAnnotation{})
		case blocksC:
			collection.docType = reflect.TypeOf(backingBlock{})
		case statusesC:
			collection.docType = reflect.TypeOf(backingStatus{})
			collection.subsidiary = true
		case constraintsC:
			collection.docType = reflect.TypeOf(backingConstraints{})
			collection.subsidiary = true
		case settingsC:
			collection.docType = reflect.TypeOf(backingSettings{})
			collection.subsidiary = true
		case openedPortsC:
			collection.docType = reflect.TypeOf(backingOpenedPorts{})
			collection.subsidiary = true
		case remoteApplicationsC:
			collection.docType = reflect.TypeOf(backingRemoteApplication{})
		case applicationOffersC:
			collection.docType = reflect.TypeOf(backingApplicationOffer{})
		default:
			panic(errors.Errorf("unknown collection %q", collName))
		}

		docType := collection.docType
		if _, ok := seenTypes[docType]; ok {
			panic(errors.Errorf("duplicate collection type %s", docType))
		}
		seenTypes[docType] = struct{}{}

		if _, ok := collectionByName[collName]; ok {
			panic(errors.Errorf("duplicate collection name %q", collName))
		}
		collectionByName[collName] = collection
	}

	return collectionByName
}
922869
func (s *Store) ChecksInStateByNodeMeta(ws memdb.WatchSet, state string, filters map[string]string) (uint64, structs.HealthChecks, error) {
	tx := s.db.Txn(false)
	defer tx.Abort()

	// Get the table index.
	idx := maxIndexTxn(tx, "nodes", "checks")

	// Query all checks if HealthAny is passed, otherwise use the index.
	var iter memdb.ResultIterator
	var err error
	if state == api.HealthAny {
		iter, err = tx.Get("checks", "status")
		if err != nil {
			return 0, nil, fmt.Errorf("failed check lookup: %s", err)
		}
	} else {
		iter, err = tx.Get("checks", "status", state)
		if err != nil {
			return 0, nil, fmt.Errorf("failed check lookup: %s", err)
		}
	}
	ws.Add(iter.WatchCh())

	return s.parseChecksByNodeMeta(tx, ws, idx, iter, filters)
}
1151989
func (d *EtcdStateDriver) WatchAll(baseKey string, rsps chan [2][]byte) error {
	watcher := d.KeysAPI.Watcher(baseKey, &client.WatcherOptions{Recursive: true})
	if watcher == nil {
		log.Errorf("etcd watch failed.")
		return errors.New("etcd watch failed")
	}

	go d.channelEtcdEvents(watcher, rsps)

	return nil
}
650649
func (s *CfgNetworkState) WatchAll(rsps chan core.WatchState) error {
	return s.StateDriver.WatchAllState(networkConfigPathPrefix, s, json.Unmarshal,
		rsps)
}
1268957
func MonitorState(cancel <-chan struct{}, conn client.Connection, req StateRequest, check func(s *State, exists bool) bool) (*State, error) {
	logger := plog.WithFields(log.Fields{
		"hostid":     req.HostID,
		"serviceid":  req.ServiceID,
		"instanceid": req.InstanceID,
	})

	basepth := "/"
	if req.PoolID != "" {
		basepth = path.Join("/pools", req.PoolID)
	}

	// Monitor the state of the object
	done := make(chan struct{})
	defer func() { close(done) }()
	for {
		// Get the current host state and set the watch
		hspth := path.Join(basepth, "/hosts", req.HostID, "instances", req.StateID())
		hsok := true
		hsdat := &HostState{}
		hsev, err := conn.GetW(hspth, hsdat, done)
		if err == client.ErrNoNode {
			hsok = false
		} else if err != nil {
			logger.WithError(err).Debug("Could not watch host state")
			return nil, &StateError{
				Request:   req,
				Operation: "watch",
				Message:   "could not watch host state",
			}
		}

		// Get the current service state and set the watch
		sspth := path.Join(basepth, "/services", req.ServiceID, req.StateID())
		ssok := true
		ssdat := &ServiceState{}
		ssev, err := conn.GetW(sspth, ssdat, done)
		if err == client.ErrNoNode {
			ssok = false
		} else if err != nil {
			logger.WithError(err).Debug("Could not watch service state")
			return nil, &StateError{
				Request:   req,
				Operation: "watch",
				Message:   "could not watch service state",
			}
		}

		cspth := path.Join(sspth, "current")
		csok := true
		cstate := &CurrentStateContainer{}
		csev, err := conn.GetW(cspth, cstate, done)
		if err == client.ErrNoNode {
			csok = false
		} else if err != nil {
			logger.WithError(err).Debug("Could not watch service status")
			return nil, &StateError{
				Request:   req,
				Operation: "watch",
				Message:   "could not watch service status",
			}
		}

		// Ensure the nodes are compatible
		if hsok != ssok || ssok != csok {
			logger.WithFields(log.Fields{
				"hoststateexists":     hsok,
				"servicestateexists":  ssok,
				"servicestatusexists": csok,
			}).Debug("Incongruent state")
			return nil, &StateError{
				Request:   req,
				Operation: "watch",
				Message:   "incongruent state",
			}
		}
		exists := hsok && ssok && csok

		// Does the state statisfy the requirements?
		state := &State{
			HostState:             *hsdat,
			ServiceState:          *ssdat,
			CurrentStateContainer: *cstate,
			HostID:                req.HostID,
			ServiceID:             req.ServiceID,
			InstanceID:            req.InstanceID,
		}

		// Only return the state if the value is true.  Return an error if the
		// state has been deleted.
		if check(state, exists) {
			return state, nil
		} else if !exists {
			logger.Debug("State does not exist")
			return nil, &StateError{
				Request:   req,
				Operation: "watch",
				Message:   "state does not exist",
			}
		}

		// Wait for something to happen
		select {
		case <-hsev:
		case <-ssev:
		case <-csev:
		case <-cancel:
			logger.Debug("Aborted state monitor")
			return nil, nil
		}

		close(done)
		done = make(chan struct{})
	}
}
1123482
func (c *Client) watchAllocations(updates chan *allocUpdates) {
	// The request and response for getting the map of allocations that should
	// be running on the Node to their AllocModifyIndex which is incremented
	// when the allocation is updated by the servers.
	req := structs.NodeSpecificRequest{
		NodeID:   c.NodeID(),
		SecretID: c.secretNodeID(),
		QueryOptions: structs.QueryOptions{
			Region:     c.Region(),
			AllowStale: true,
		},
	}
	var resp structs.NodeClientAllocsResponse

	// The request and response for pulling down the set of allocations that are
	// new, or updated server side.
	allocsReq := structs.AllocsGetRequest{
		QueryOptions: structs.QueryOptions{
			Region:     c.Region(),
			AllowStale: true,
		},
	}
	var allocsResp structs.AllocsGetResponse

OUTER:
	for {
		// Get the allocation modify index map, blocking for updates. We will
		// use this to determine exactly what allocations need to be downloaded
		// in full.
		resp = structs.NodeClientAllocsResponse{}
		err := c.RPC("Node.GetClientAllocs", &req, &resp)
		if err != nil {
			// Shutdown often causes EOF errors, so check for shutdown first
			select {
			case <-c.shutdownCh:
				return
			default:
			}

			// COMPAT: Remove in 0.6. This is to allow the case in which the
			// servers are not fully upgraded before the clients register. This
			// can cause the SecretID to be lost
			if strings.Contains(err.Error(), "node secret ID does not match") {
				c.logger.Debug("secret mismatch; re-registering node", "error", err)
				c.retryRegisterNode()
			} else if err != noServersErr {
				c.logger.Error("error querying node allocations", "error", err)
			}
			retry := c.retryIntv(getAllocRetryIntv)
			select {
			case <-c.rpcRetryWatcher():
				continue
			case <-time.After(retry):
				continue
			case <-c.shutdownCh:
				return
			}
		}

		// Check for shutdown
		select {
		case <-c.shutdownCh:
			return
		default:
		}

		// Filter all allocations whose AllocModifyIndex was not incremented.
		// These are the allocations who have either not been updated, or whose
		// updates are a result of the client sending an update for the alloc.
		// This lets us reduce the network traffic to the server as we don't
		// need to pull all the allocations.
		var pull []string
		filtered := make(map[string]struct{})
		var pullIndex uint64
		for allocID, modifyIndex := range resp.Allocs {
			// Pull the allocation if we don't have an alloc runner for the
			// allocation or if the alloc runner requires an updated allocation.
			//XXX Part of Client alloc index tracking exp
			c.allocLock.RLock()
			currentAR, ok := c.allocs[allocID]
			c.allocLock.RUnlock()

			// Ignore alloc updates for allocs that are invalid because of initialization errors
			_, isInvalid := c.invalidAllocs[allocID]

			if (!ok || modifyIndex > currentAR.Alloc().AllocModifyIndex) && !isInvalid {
				// Only pull allocs that are required. Filtered
				// allocs might be at a higher index, so ignore
				// it.
				if modifyIndex > pullIndex {
					pullIndex = modifyIndex
				}
				pull = append(pull, allocID)
			} else {
				filtered[allocID] = struct{}{}
			}
		}

		// Pull the allocations that passed filtering.
		allocsResp.Allocs = nil
		var pulledAllocs map[string]*structs.Allocation
		if len(pull) != 0 {
			// Pull the allocations that need to be updated.
			allocsReq.AllocIDs = pull
			allocsReq.MinQueryIndex = pullIndex - 1
			allocsResp = structs.AllocsGetResponse{}
			if err := c.RPC("Alloc.GetAllocs", &allocsReq, &allocsResp); err != nil {
				c.logger.Error("error querying updated allocations", "error", err)
				retry := c.retryIntv(getAllocRetryIntv)
				select {
				case <-c.rpcRetryWatcher():
					continue
				case <-time.After(retry):
					continue
				case <-c.shutdownCh:
					return
				}
			}

			// Ensure that we received all the allocations we wanted
			pulledAllocs = make(map[string]*structs.Allocation, len(allocsResp.Allocs))
			for _, alloc := range allocsResp.Allocs {
				pulledAllocs[alloc.ID] = alloc
			}

			for _, desiredID := range pull {
				if _, ok := pulledAllocs[desiredID]; !ok {
					// We didn't get everything we wanted. Do not update the
					// MinQueryIndex, sleep and then retry.
					wait := c.retryIntv(2 * time.Second)
					select {
					case <-time.After(wait):
						// Wait for the server we contact to receive the
						// allocations
						continue OUTER
					case <-c.shutdownCh:
						return
					}
				}
			}

			// Check for shutdown
			select {
			case <-c.shutdownCh:
				return
			default:
			}
		}

		c.logger.Debug("updated allocations", "index", resp.Index,
			"total", len(resp.Allocs), "pulled", len(allocsResp.Allocs), "filtered", len(filtered))

		// Update the query index.
		if resp.Index > req.MinQueryIndex {
			req.MinQueryIndex = resp.Index
		}

		// Push the updates.
		update := &allocUpdates{
			filtered:      filtered,
			pulled:        pulledAllocs,
			migrateTokens: resp.MigrateTokens,
		}

		select {
		case updates <- update:
		case <-c.shutdownCh:
			return
		}
	}
}