// Watch a given file path for a particular set of notifications (FSN_MODIFY etc.)
169927
func (w *Watcher) WatchFlags(path string, flags uint32) error {
	w.fsnmut.Lock()
	w.fsnFlags[path] = flags
	w.fsnmut.Unlock()
	return w.watch(path)
}
928236
func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
	var msg *ReceivedMessage

	fs.mutex.RLock()
	defer fs.mutex.RUnlock()

	candidates := fs.getWatchersByTopic(env.Topic)
	for _, watcher := range candidates {
		if p2pMessage && !watcher.AllowP2P {
			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcher.id))
			continue
		}

		var match bool
		if msg != nil {
			match = watcher.MatchMessage(msg)
		} else {
			match = watcher.MatchEnvelope(env)
			if match {
				msg = env.Open(watcher)
				if msg == nil {
					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcher.id)
				}
			} else {
				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcher.id)
			}
		}

		if match && msg != nil {
			log.Trace("processing message: decrypted", "hash", env.Hash().Hex())
			if watcher.Src == nil || IsPubKeyEqual(msg.Src, watcher.Src) {
				watcher.Trigger(msg)
			}
		}
	}
}
989695
func (i *inotify) Rewatch(path string, _, newevent Event) error {
	return i.watch(path, newevent)
}
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)
}
648214
func (m *Machine) Watch() NotifyWatcher {
	return newEntityWatcher(m.st, machinesC, m.doc.DocID)
}
1330119
func AddRecursiveWatch(watcher *fsnotify.Watcher, path string) error {
	file, err := os.Stat(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return fmt.Errorf("error introspecting path %s: %v", path, err)
	}
	if !file.IsDir() {
		return nil
	}

	folders, err := getSubFolders(path)
	for _, v := range folders {
		klog.V(5).Infof("adding watch on path %s", v)
		err = watcher.Add(v)
		if err != nil {
			// "no space left on device" issues are usually resolved via
			// $ sudo sysctl fs.inotify.max_user_watches=65536
			return fmt.Errorf("error adding watcher for path %s: %v", v, err)
		}
	}
	return nil
}
1347875
func (c *Conn) Watch(ctx context.Context, filePath string) (*topo.WatchData, <-chan *topo.WatchData, topo.CancelFunc) {
	c.factory.mu.Lock()
	defer c.factory.mu.Unlock()

	if c.factory.err != nil {
		return &topo.WatchData{Err: c.factory.err}, nil, nil
	}

	n := c.factory.nodeByPath(c.cell, filePath)
	if n == nil {
		return &topo.WatchData{Err: topo.NewError(topo.NoNode, filePath)}, nil, nil
	}
	if n.contents == nil {
		// it's a directory
		return &topo.WatchData{Err: fmt.Errorf("cannot watch directory %v in cell %v", filePath, c.cell)}, nil, nil
	}
	current := &topo.WatchData{
		Contents: n.contents,
		Version:  NodeVersion(n.version),
	}

	notifications := make(chan *topo.WatchData, 100)
	watchIndex := nextWatchIndex
	nextWatchIndex++
	n.watches[watchIndex] = notifications

	cancel := func() {
		// This function can be called at any point, so we first need
		// to make sure the watch is still valid.
		c.factory.mu.Lock()
		defer c.factory.mu.Unlock()

		n := c.factory.nodeByPath(c.cell, filePath)
		if n == nil {
			return
		}

		if w, ok := n.watches[watchIndex]; ok {
			delete(n.watches, watchIndex)
			w <- &topo.WatchData{Err: topo.NewError(topo.Interrupted, "watch")}
			close(w)
		}
	}
	return current, notifications, cancel
}
932511
func (m *RWatcher) watchRecursive(path string, unWatch bool) error {
	err := filepath.Walk(path, func(walkPath string, fi os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if fi.IsDir() {
			if unWatch {
				if err = m.fsnotify.Remove(walkPath); err != nil {
					return err
				}
			} else {
				if err = m.fsnotify.Add(walkPath); err != nil {
					return err
				}
			}
		}
		return nil
	})
	return err
}
728330
func (w *Watcher) WatchedFiles() map[string]os.FileInfo {
	w.mu.Lock()
	defer w.mu.Unlock()

	files := make(map[string]os.FileInfo)
	for k, v := range w.files {
		files[k] = v
	}

	return files
}
422206
func startWatcher(paths []string, ch chan int) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		beeLogger.Log.Fatalf("Could not start the watcher: %v", err)
	}
	defer watcher.Close()

	// Feed the paths to the watcher
	for _, path := range paths {
		if err := watcher.Add(path); err != nil {
			beeLogger.Log.Fatalf("Could not set a watch on path: %v", err)
		}
	}

	for {
		select {
		case evt := <-watcher.Events:
			build := true
			if filepath.Ext(evt.Name) != ".go" {
				continue
			}

			mt := utils.GetFileModTime(evt.Name)
			if t := eventsModTime[evt.Name]; mt == t {
				build = false
			}
			eventsModTime[evt.Name] = mt

			if build {
				go func() {
					if verbose {
						utils.Notify("Rebuilding application with the new changes", "bee")
					}

					// Wait 1s before re-build until there is no file change
					scheduleTime := time.Now().Add(1 * time.Second)
					time.Sleep(time.Until(scheduleTime))
					_, err := buildDebug()
					if err != nil {
						utils.Notify("Build Failed: "+err.Error(), "bee")
					} else {
						ch <- 0 // Notify listeners
					}
				}()
			}
		case err := <-watcher.Errors:
			if err != nil {
				ch <- -1
			}
		}
	}
}
1206452
func (c *FakeRoutes) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(routesResource, c.ns, opts))

}
1018813
func watchForChanges(configPath string, stopCh chan struct{}) error {
	configPath, err := filepath.Abs(configPath)
	if err != nil {
		return err
	}

	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return err
	}

	// Watch all symlinks for changes
	p := configPath
	maxdepth := 100
	for depth := 0; depth < maxdepth; depth++ {
		if err := watcher.Add(p); err != nil {
			return err
		}
		klog.V(2).Infof("Watching config file %s for changes", p)

		stat, err := os.Lstat(p)
		if err != nil {
			return err
		}

		// configmaps are usually symlinks
		if stat.Mode()&os.ModeSymlink > 0 {
			p, err = filepath.EvalSymlinks(p)
			if err != nil {
				return err
			}
		} else {
			break
		}
	}

	go func() {
		for {
			select {
			case <-stopCh:
			case event, ok := <-watcher.Events:
				if !ok {
					return
				}
				klog.V(2).Infof("Configuration file %s changed, exiting...", event.Name)
				close(stopCh)
				return
			case err, ok := <-watcher.Errors:
				if !ok {
					return
				}
				klog.V(4).Infof("fsnotify error %v", err)
			}
		}
	}()
	return nil
}
406278
func (w *Watcher) watch(path string) error {
	w.ewmut.Lock()
	w.externalWatches[path] = true
	w.ewmut.Unlock()
	return w.addWatch(path, sys_NOTE_ALLEVENTS)
}
1193863
func (r *readdcw) rewatch(path string, oldevent, newevent uint32, recursive bool) (err error) {
	if Event(newevent)&^(All|fileNotifyChangeAll) != 0 {
		return errors.New("notify: unknown event")
	}
	var wd *watched
	r.Lock()
	defer r.Unlock()
	if wd, err = r.nonStateWatchedLocked(path); err != nil {
		return
	}
	if wd.filter&(onlyNotifyChanges|onlyNGlobalEvents) != oldevent {
		panic(`notify: windows re-watcher logic error`)
	}
	wd.filter = stateRewatch | newevent
	wd.recursive, recursive = recursive, wd.recursive
	if err = wd.closeHandle(); err != nil {
		wd.filter = oldevent
		wd.recursive = recursive
		return
	}
	return
}
1097345
func (api *API) Watch() params.NotifyWatchResult {
	watch := api.backend.WatchForMigration()
	if _, ok := <-watch.Changes(); ok {
		return params.NotifyWatchResult{
			NotifyWatcherId: api.resources.Register(watch),
		}
	}
	return params.NotifyWatchResult{
		Error: common.ServerError(watcher.EnsureErr(watch)),
	}
}
273557
func WatchTokenFile(tokenfile string, done <-chan interface{}) error {
	log := log.WithFields(logrus.Fields{
		"tokenfile": tokenfile,
	})

	loadToken := func() {
		log := log.WithFields(logrus.Fields{
			"tokenfile": tokenfile,
		})

		if err := LoadTokenFile(tokenfile); err != nil {
			log.WithError(err).Warn("Unable to load authentication token from file.")
		} else {
			log.Debug("Updated authentication token from disk")
		}
	}

	// An initial token load without any file changes
	loadToken()

	// Now watch for changes
	filechangechan, err := NotifyOnChange(tokenfile, fsnotify.Write|fsnotify.Create, done)
	if err != nil {
		return err
	}
	for _ = range filechangechan {
		loadToken()
	}
	return nil
}
897131
func (u *Unit) Watch() NotifyWatcher {
	return newEntityWatcher(u.st, unitsC, u.doc.DocID)
}