// runWatch handles any update events from the cache.Notify until the
// config watch is shut down.
974751
func (s *serviceConfigWatch) runWatch() {
	firstRun := true
	for {
		select {
		case <-s.ctx.Done():
			return
		case event := <-s.updateCh:
			if err := s.handleUpdate(event, false, firstRun); err != nil {
				s.agent.logger.Printf("[ERR] agent.manager: error handling service update: %v", err)
			}
			firstRun = false
		}
	}
}
432217
func (e *events) Watch(opts api.ListOptions) (watch.Interface, error) {
	return e.client.Get().
		Prefix("watch").
		NamespaceIfScoped(e.namespace, len(e.namespace) > 0).
		Resource("events").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
61093
func (sr *SimpleRedis) watch(done chan struct{}) {
	sr.log.Info("==> Redis DB watcher started")

	var count int8

	for {
		select {
		case r, ok := <-sr.respChan:
			count++
			if !ok {
				sr.log.Info("==> Redis DB watcher closed")
				return
			}
			switch r.GetChangeType() {
			case datasync.Put:
				sr.log.Debugf("==> Redis watcher: received 'put' event: key: %s, value: %s", r.GetKey(), string(r.GetValue()))
			case datasync.Delete:
				sr.log.Debugf("==> Redis watcher: received 'delete' event: key: %s", r.GetKey())
			}
			if count == 12 {
				sr.log.Info("All expected events were received by watcher")
				done <- struct{}{}
			}
		}
	}
}
993664
func (w *userProjectWatcher) Watch() {
	defer close(w.outgoing)
	defer func() {
		// when the watch ends, always remove the watcher from the cache to avoid leaking.
		w.authCache.RemoveWatcher(w)
	}()
	defer utilruntime.HandleCrash()

	// start by emitting all the `initialProjects`
	for i := range w.initialProjects {
		// keep this check here to sure we don't keep this open in the case of failures
		select {
		case err := <-w.cacheError:
			w.emit(makeErrorEvent(err))
			return
		default:
		}

		w.emit(watch.Event{
			Type:   watch.Added,
			Object: projectutil.ConvertNamespaceFromExternal(&w.initialProjects[i]),
		})
	}

	for {
		select {
		case err := <-w.cacheError:
			w.emit(makeErrorEvent(err))
			return

		case <-w.userStop:
			return

		case event := <-w.cacheIncoming:
			if curLen := int64(len(w.cacheIncoming)); watchChannelHWM.Update(curLen) {
				// Monitor if this gets backed up, and how much.
				klog.V(2).Infof("watch: %v objects queued in project cache watching channel.", curLen)
			}

			w.emit(event)
		}
	}
}
985856
func (c *FakeKeysets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(keysetsResource, c.ns, opts))

}
909868
func (c *FakeControllerRevisions) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(controllerrevisionsResource, c.ns, opts))

}
273854
func (w *checkWatcher) Watch(allocID, taskName, checkID string, check *structs.ServiceCheck, restarter TaskRestarter) {
	if !check.TriggersRestarts() {
		// Not watched, noop
		return
	}

	c := &checkRestart{
		allocID:        allocID,
		taskName:       taskName,
		checkID:        checkID,
		checkName:      check.Name,
		taskKey:        fmt.Sprintf("%s%s", allocID, taskName), // unique task ID
		task:           restarter,
		interval:       check.Interval,
		grace:          check.CheckRestart.Grace,
		graceUntil:     time.Now().Add(check.CheckRestart.Grace),
		timeLimit:      check.Interval * time.Duration(check.CheckRestart.Limit-1),
		ignoreWarnings: check.CheckRestart.IgnoreWarnings,
		logger:         w.logger.With("alloc_id", allocID, "task", taskName, "check", check.Name),
	}

	update := checkWatchUpdate{
		checkID:      checkID,
		checkRestart: c,
	}

	select {
	case w.checkUpdateCh <- update:
		// sent watch
	case <-w.done:
		// exited; nothing to do
	}
}
52723
func (cache *snapshotCache) cancelWatch(nodeID string, watchID int64) func() {
	return func() {
		// uses the cache mutex
		cache.mu.Lock()
		defer cache.mu.Unlock()
		if info, ok := cache.status[nodeID]; ok {
			info.mu.Lock()
			delete(info.watches, watchID)
			info.mu.Unlock()
		}
	}
}
702744
func (mr *ModRunner) Run() error {
	for {
		modchan := make(chan *moddwatch.Mod, 1024)
		err := mr.runOnChan(modchan, func() {})
		if err != nil {
			return err
		}
	}
}
193210
func (c *FakeMiddlewares) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(middlewaresResource, c.ns, opts))

}
317158
func (a *AgentEntityWatcher) Watch(args params.Entities) (params.NotifyWatchResults, error) {
	result := params.NotifyWatchResults{
		Results: make([]params.NotifyWatchResult, len(args.Entities)),
	}
	if len(args.Entities) == 0 {
		return result, nil
	}
	canWatch, err := a.getCanWatch()
	if err != nil {
		return params.NotifyWatchResults{}, errors.Trace(err)
	}
	for i, entity := range args.Entities {
		tag, err := names.ParseTag(entity.Tag)
		if err != nil {
			result.Results[i].Error = ServerError(ErrPerm)
			continue
		}
		err = ErrPerm
		watcherId := ""
		if canWatch(tag) {
			watcherId, err = a.watchEntity(tag)
		}
		result.Results[i].NotifyWatcherId = watcherId
		result.Results[i].Error = ServerError(err)
	}
	return result, nil
}
269057
func (f FakeWatcher) Watch(api.ListOptions) (watch.Interface, error) {
	return f.IFace, f.Err
}
127764
func (t *Transaction) Watch(key ...string) error {
	if len(key) == 0 {
		return nil
	}
	args := make([]interface{}, len(key))
	for i := range key {
		args[i] = key[i]
	}
	_, err := NewCommand("WATCH", args...).Run(t.conn)
	return err
}
1204511
func (w *Watcher) watch(path string) error {
	return w.AddWatch(path, sys_FS_ALL_EVENTS)
}
871962
func (ctrl *ApplicationController) Run(ctx context.Context, statusProcessors int, operationProcessors int) {
	defer runtime.HandleCrash()
	defer ctrl.appRefreshQueue.ShutDown()

	go ctrl.appInformer.Run(ctx.Done())
	go ctrl.projInformer.Run(ctx.Done())
	go ctrl.watchSettings(ctx)

	if !cache.WaitForCacheSync(ctx.Done(), ctrl.appInformer.HasSynced, ctrl.projInformer.HasSynced) {
		log.Error("Timed out waiting for caches to sync")
		return
	}

	go ctrl.stateCache.Run(ctx)
	go func() { errors.CheckError(ctrl.metricsServer.ListenAndServe()) }()

	for i := 0; i < statusProcessors; i++ {
		go wait.Until(func() {
			for ctrl.processAppRefreshQueueItem() {
			}
		}, time.Second, ctx.Done())
	}

	for i := 0; i < operationProcessors; i++ {
		go wait.Until(func() {
			for ctrl.processAppOperationQueueItem() {
			}
		}, time.Second, ctx.Done())
	}

	<-ctx.Done()
}
622845
func (mr *MockDeploymentInterfaceMockRecorder) Watch(arg0 interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockDeploymentInterface)(nil).Watch), arg0)
}
479628
func (c *FakeStorageClasses) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(storageclassesResource, opts))
}