// Rewatch implements Watcher interface.
//
// TODO(rjeczalik): This is a naive hack. Rewrite might help.
12896
func (t *trg) Rewatch(p string, _, e Event) error {
	fi, err := os.Stat(p)
	if err != nil {
		return err
	}
	t.Lock()
	if err = t.unwatch(p, fi); err == nil {
		// TODO(rjeczalik): If watch fails then we leave trigger in inconsistent
		// state. Handle? Panic? Native version of rewatch?
		err = t.watch(p, e, fi)
	}
	t.Unlock()
	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)
	}
}
1010432
func (adapter *Adapter) Watch(resyncName string, changeChan chan datasync.ChangeEvent,
	resyncChan chan datasync.ResyncEvent, keyPrefixes ...string) (datasync.WatchRegistration, error) {

	logrus.DefaultLogger().Debug("REST KeyValProtoWatcher WatchData ", resyncName, " ", keyPrefixes)

	for _, keyPrefix := range keyPrefixes {
		adapter.registerHTTPHandler(keyPrefix+"{suffix}", adapter.putMessage, "PUT")
		adapter.registerHTTPHandler(keyPrefix+"{suffix}", adapter.delMessage, "DELETE")
		//TODO adapter.registerHTTPHandler(keyPrefix + "{suffix}", getMessage, "GET")
		//TODO httpmux.RegisterHTTPHandler("/vpprestcon/resync", putResync, "PUT")
	}

	return adapter.base.Watch(resyncName, changeChan, resyncChan, keyPrefixes...)
}
222725
func Watch(programs []Program, restartDelay time.Duration, log Log) error {
	// Terminate any existing programs that we are supposed to monitor
	terminateExisting(programs, log)

	// Start monitoring all the programs
	watchPrograms(programs, restartDelay, log)
	return nil
}
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
}
940188
func (c *FakeImageStreams) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(imagestreamsResource, c.ns, opts))

}
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
}
625834
func (m *Machine) Watch() (watcher.NotifyWatcher, error) {
	return common.Watch(m.st.facade, "Watch", m.tag)
}
1206452
func (c *FakeRoutes) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(routesResource, c.ns, opts))

}
511388
func (c *namespaces) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Resource("namespaces").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
1364525
func (ta KVProtoWatchers) Watch(resyncName string, changeChan chan ChangeEvent,
	resyncChan chan ResyncEvent, keyPrefixes ...string) (WatchRegistration, error) {

	var registrations []WatchRegistration

	for _, adapter := range ta {
		watcherReg, err := adapter.Watch(resyncName, changeChan, resyncChan, keyPrefixes...)
		if err != nil {
			return nil, err
		}

		if watcherReg != nil {
			registrations = append(registrations, watcherReg)
		}
	}

	return &AggregatedRegistration{
		Registrations: registrations,
	}, nil
}
1105736
func (t *nonrecursiveTree) Watch(path string, c chan<- EventInfo, events ...Event) error {
	if c == nil {
		panic("notify: Watch using nil channel")
	}
	// Expanding with empty event set is a nop.
	if len(events) == 0 {
		return nil
	}
	path, isrec, err := cleanpath(path)
	if err != nil {
		return err
	}
	eset := joinevents(events)
	t.rw.Lock()
	defer t.rw.Unlock()
	nd := t.root.Add(path)
	if isrec {
		return t.watchrec(nd, c, eset|recursive)
	}
	return t.watch(nd, c, eset)
}
792594
func (m *MockBaseWatcher) Watch(arg0 string, arg1 interface{}, arg2 chan<- watcher.Change) {
	m.ctrl.Call(m, "Watch", arg0, arg1, arg2)
}
648214
func (m *Machine) Watch() NotifyWatcher {
	return newEntityWatcher(m.st, machinesC, m.doc.DocID)
}
1003617
func (e *events) Watch(opts api.ListOptions) (watch.Interface, error) {
	return e.client.Get().
		Prefix("watch").
		Namespace(e.namespace).
		Resource("events").
		VersionedParams(&opts, api.ParameterCodec).
		Watch()
}
832446
func (s *Server) Watch(ctx context.Context, filePath string) (*topo.WatchData, <-chan *topo.WatchData, topo.CancelFunc) {
	nodePath := path.Join(s.root, filePath)

	// Get the initial version of the file
	initial, err := s.cli.Get(ctx, nodePath)
	if err != nil {
		// Generic error.
		return &topo.WatchData{Err: convertError(err, nodePath)}, nil, nil
	}
	if len(initial.Kvs) != 1 {
		// Node doesn't exist.
		return &topo.WatchData{Err: topo.NewError(topo.NoNode, nodePath)}, nil, nil
	}
	wd := &topo.WatchData{
		Contents: initial.Kvs[0].Value,
		Version:  EtcdVersion(initial.Kvs[0].ModRevision),
	}

	// Create a context, will be used to cancel the watch.
	watchCtx, watchCancel := context.WithCancel(context.Background())

	// Create the Watcher.  We start watching from the response we
	// got, not from the file original version, as the server may
	// not have that much history.
	watcher := s.cli.Watch(watchCtx, nodePath, clientv3.WithRev(initial.Header.Revision))
	if watcher == nil {
		return &topo.WatchData{Err: vterrors.Errorf(vtrpc.Code_INVALID_ARGUMENT, "Watch failed")}, nil, nil
	}

	// Create the notifications channel, send updates to it.
	notifications := make(chan *topo.WatchData, 10)
	go func() {
		defer close(notifications)

		var currVersion = initial.Header.Revision
		var watchRetries int
		for {
			select {

			case <-watchCtx.Done():
				// This includes context cancelation errors.
				notifications <- &topo.WatchData{
					Err: convertError(watchCtx.Err(), nodePath),
				}
				return
			case wresp, ok := <-watcher:
				if !ok {
					if watchRetries > 10 {
						time.Sleep(time.Duration(watchRetries) * time.Second)
					}
					watchRetries++
					newWatcher := s.cli.Watch(watchCtx, nodePath, clientv3.WithRev(currVersion))
					if newWatcher == nil {
						log.Warningf("watch %v failed and get a nil channel returned, currVersion: %v", nodePath, currVersion)
					} else {
						watcher = newWatcher
					}
					continue
				}

				watchRetries = 0

				if wresp.Canceled {
					// Final notification.
					notifications <- &topo.WatchData{
						Err: convertError(wresp.Err(), nodePath),
					}
					return
				}

				currVersion = wresp.Header.GetRevision()

				for _, ev := range wresp.Events {
					switch ev.Type {
					case mvccpb.PUT:
						notifications <- &topo.WatchData{
							Contents: ev.Kv.Value,
							Version:  EtcdVersion(ev.Kv.Version),
						}
					case mvccpb.DELETE:
						// Node is gone, send a final notice.
						notifications <- &topo.WatchData{
							Err: topo.NewError(topo.NoNode, nodePath),
						}
						return
					default:
						notifications <- &topo.WatchData{
							Err: vterrors.Errorf(vtrpc.Code_INTERNAL, "unexpected event received: %v", ev),
						}
						return
					}
				}
			}
		}
	}()

	return wd, notifications, topo.CancelFunc(watchCancel)
}