// Watch the prefix for any events and send a 'true' if we should attempt grab leader
768178
func (s *EtcdElection) Watch(ctx context.Context, startIndex uint64) chan bool {
	results := make(chan bool)
	var cancel atomic.Value
	var stop int32

	go func() {
		select {
		case <-ctx.Done():
			cancel.Load().(context.CancelFunc)()
			return
		}
	}()

	// Create our initial watcher
	watcher := s.api.Watcher(s.conf.Election, nil)

	s.wg.Loop(func() bool {
		// this ensures we exit properly if the watcher isn't connected to etcd
		if atomic.LoadInt32(&stop) == 1 {
			return false
		}

		ctx, c := context.WithTimeout(context.Background(), time.Second*10)
		cancel.Store(c)
		resp, err := watcher.Next(ctx)
		if err != nil {
			if err == context.Canceled {
				close(results)
				return false
			}

			if cErr, ok := err.(etcd.Error); ok {
				if cErr.Code == etcd.ErrorCodeEventIndexCleared {
					logrus.WithField("category", "election").
						Infof("EtcdElection %s - new index is %d",
							err, cErr.Index+1)

					// Re-create the watcher with a newer index until we catch up
					watcher = s.api.Watcher(s.conf.Election, nil)
					return true
				}
			}
			if err != context.DeadlineExceeded {
				logrus.WithField("category", "election").
					Errorf("EtcdElection etcd error: %s", err)
				time.Sleep(time.Second * 10)
			}
		} else {
			logrus.WithField("category", "election").
				Debugf("EtcdElection etcd event: %+v\n", resp)
			results <- true
		}
		return true
	})
	return results
}
511388
func (c *namespaces) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Resource("namespaces").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
22013
func (s *Sender) Watch(key string, ctx context.Context) <-chan *Event {
	s.Lock()
	defer s.Unlock()

	log.Infof("A new receiver watching to %s", key)
	r := &Receiver{
		key:     key,
		ctx:     ctx,
		ch:      make(chan *Event, 1),
		counter: 2,
	}
	s.receivers = append(s.receivers, r)
	return (<-chan *Event)(r.ch)
}
1206452
func (c *FakeRoutes) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(routesResource, c.ns, opts))

}
290696
func (c *clientWrapper) WatchAll(namespaces k8s.Namespaces, stopCh <-chan struct{}) (<-chan interface{}, error) {
	eventCh := make(chan interface{}, 1)
	eventHandler := c.newResourceEventHandler(eventCh)

	if len(namespaces) == 0 {
		namespaces = k8s.Namespaces{metav1.NamespaceAll}
		c.isNamespaceAll = true
	}
	c.watchedNamespaces = namespaces

	for _, ns := range namespaces {
		factoryCrd := externalversions.NewSharedInformerFactoryWithOptions(c.csCrd, resyncPeriod, externalversions.WithNamespace(ns))
		factoryCrd.Traefik().V1alpha1().IngressRoutes().Informer().AddEventHandler(eventHandler)
		factoryCrd.Traefik().V1alpha1().Middlewares().Informer().AddEventHandler(eventHandler)

		factoryKube := informers.NewFilteredSharedInformerFactory(c.csKube, resyncPeriod, ns, nil)
		factoryKube.Extensions().V1beta1().Ingresses().Informer().AddEventHandler(eventHandler)
		factoryKube.Core().V1().Services().Informer().AddEventHandler(eventHandler)
		factoryKube.Core().V1().Endpoints().Informer().AddEventHandler(eventHandler)

		c.factoriesCrd[ns] = factoryCrd
		c.factoriesKube[ns] = factoryKube
	}

	for _, ns := range namespaces {
		c.factoriesCrd[ns].Start(stopCh)
		c.factoriesKube[ns].Start(stopCh)
	}

	for _, ns := range namespaces {
		for t, ok := range c.factoriesCrd[ns].WaitForCacheSync(stopCh) {
			if !ok {
				return nil, fmt.Errorf("timed out waiting for controller caches to sync %s in namespace %q", t.String(), ns)
			}
		}

		for t, ok := range c.factoriesKube[ns].WaitForCacheSync(stopCh) {
			if !ok {
				return nil, fmt.Errorf("timed out waiting for controller caches to sync %s in namespace %q", t.String(), ns)
			}
		}
	}

	// Do not wait for the Secrets store to get synced since we cannot rely on
	// users having granted RBAC permissions for this object.
	// https://github.com/containous/traefik/issues/1784 should improve the
	// situation here in the future.
	for _, ns := range namespaces {
		c.factoriesKube[ns].Core().V1().Secrets().Informer().AddEventHandler(eventHandler)
		c.factoriesKube[ns].Start(stopCh)
	}

	return eventCh, nil
}
135793
func (i *inotify) watch(path string, e Event) (err error) {
	if e&^(All|Event(unix.IN_ALL_EVENTS)) != 0 {
		return errors.New("notify: unknown event")
	}
	if err = i.lazyinit(); err != nil {
		return
	}
	iwd, err := unix.InotifyAddWatch(int(i.fd), path, encode(e))
	if err != nil {
		return
	}
	i.RLock()
	wd := i.m[int32(iwd)]
	i.RUnlock()
	if wd == nil {
		i.Lock()
		if i.m[int32(iwd)] == nil {
			i.m[int32(iwd)] = &watched{path: path, mask: uint32(e)}
		}
		i.Unlock()
	} else {
		i.Lock()
		wd.mask = uint32(e)
		i.Unlock()
	}
	return nil
}
466466
func (plugin *ExamplePlugin) watchChanges(x datasync.ProtoWatchResp) {
	message := &ipsec.TunnelInterfaces{}
	err := x.GetValue(message)
	if err == nil {
		plugin.Log.Infof("Got watch message %v", message)
	}
}
439383
func keyPrefixWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	var prefix string
	if err := assignValue(params, "prefix", &prefix); err != nil {
		return nil, err
	}
	if prefix == "" {
		return nil, fmt.Errorf("Must specify a single prefix to watch")
	}
	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		kv := p.client.KV()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		pairs, meta, err := kv.List(prefix, &opts)
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), pairs, err
	}
	return fn, nil
}
1166774
func (ww *WidgetWatchers) Watch(handler tcell.EventHandler) {
	if ww.watchers == nil {
		ww.watchers = make(map[tcell.EventHandler]struct{})
	}
	ww.watchers[handler] = struct{}{}
}
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()
}
816939
func eventWatch(params map[string]interface{}) (WatcherFunc, error) {
	// The stale setting doesn't apply to events.

	var name string
	if err := assignValue(params, "name", &name); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		event := p.client.Event()
		opts := makeQueryOptionsWithContext(p, false)
		defer p.cancelFunc()
		events, meta, err := event.List(name, &opts)
		if err != nil {
			return nil, nil, err
		}

		// Prune to only the new events
		for i := 0; i < len(events); i++ {
			if WaitIndexVal(event.IDToIndex(events[i].ID)).Equal(p.lastParamVal) {
				events = events[i+1:]
				break
			}
		}
		return WaitIndexVal(meta.LastIndex), events, err
	}
	return fn, nil
}
1204511
func (w *Watcher) watch(path string) error {
	return w.AddWatch(path, sys_FS_ALL_EVENTS)
}
215762
func (i *Info) Watch(resourceVersion string) (watch.Interface, error) {
	return NewHelper(i.Client, i.Mapping).WatchSingle(i.Namespace, i.Name, resourceVersion)
}
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)
	}
}
1276126
func (c *cacheWatcher) sendWatchCacheEvent(event *watchCacheEvent) {
	watchEvent := c.convertToWatchEvent(event)
	if watchEvent == nil {
		// Watcher is not interested in that object.
		return
	}

	// We need to ensure that if we put event X to the c.result, all
	// previous events were already put into it before, no matter whether
	// c.done is close or not.
	// Thus we cannot simply select from c.done and c.result and this
	// would give us non-determinism.
	// At the same time, we don't want to block infinitely on putting
	// to c.result, when c.done is already closed.

	// This ensures that with c.done already close, we at most once go
	// into the next select after this. With that, no matter which
	// statement we choose there, we will deliver only consecutive
	// events.
	select {
	case <-c.done:
		return
	default:
	}

	select {
	case c.result <- *watchEvent:
	case <-c.done:
	}
}
27326
func watchInternal(log logging.Logger, watcher clientv3.Watcher, closeCh chan string, prefix string, resp func(keyval.BytesWatchResp)) error {
	ctx, cancel := context.WithCancel(context.Background())
	recvChan := watcher.Watch(ctx, prefix, clientv3.WithPrefix(), clientv3.WithPrevKV())

	go func(registeredKey string) {
		var compactRev int64
		for {
			select {
			case wresp, ok := <-recvChan:
				if !ok {
					log.WithField("prefix", prefix).Warn("Watch recv channel was closed")
					if compactRev != 0 {
						recvChan = watcher.Watch(context.Background(), prefix,
							clientv3.WithPrefix(), clientv3.WithPrevKV(), clientv3.WithRev(compactRev))
						log.WithFields(logging.Fields{
							"prefix": prefix,
							"rev":    compactRev,
						}).Warn("Watch recv channel was re-created")
						compactRev = 0
						continue
					}
					return
				}
				if wresp.Canceled {
					log.WithField("prefix", prefix).Warn("Watch was canceled")
				}
				err := wresp.Err()
				if err != nil {
					log.WithFields(logging.Fields{
						"prefix": prefix,
						"err":    err,
					}).Warn("Watch returned error")
				}
				compactRev = wresp.CompactRevision
				if compactRev != 0 {
					log.WithFields(logging.Fields{
						"prefix": prefix,
						"rev":    compactRev,
					}).Warn("Watched data were compacted ")
				}
				for _, ev := range wresp.Events {
					handleWatchEvent(log, resp, ev)
				}

			case closeVal, ok := <-closeCh:
				if !ok || closeVal == registeredKey {
					cancel()
					log.WithField("prefix", prefix).Debug("Watch ended")
					return
				}
			}
		}
	}(prefix)

	return nil
}
1046483
func (c *Client) WatchAll() (*AllWatcher, error) {
	var info params.AllWatcherId
	if err := c.facade.FacadeCall("WatchAll", nil, &info); err != nil {
		return nil, err
	}
	return NewAllWatcher(c.st, &info.AllWatcherId), nil
}