// RecursiveWatch implements notify.RecursiveWatcher interface.
493225
func (r *readdcw) RecursiveWatch(path string, event Event) error {
	return r.watch(path, event, true)
}
83374
func (f *fsClient) Watch(params watchParams) (*watchObject, *probe.Error) {
	eventChan := make(chan EventInfo)
	errorChan := make(chan *probe.Error)
	doneChan := make(chan bool)
	// Make the channel buffered to ensure no event is dropped. Notify will drop
	// an event if the receiver is not able to keep up the sending pace.
	in, out := PipeChan(1000)

	var fsEvents []notify.Event
	for _, event := range params.events {
		switch event {
		case "put":
			fsEvents = append(fsEvents, EventTypePut...)
		case "delete":
			fsEvents = append(fsEvents, EventTypeDelete...)
		case "get":
			fsEvents = append(fsEvents, EventTypeGet...)
		default:
			return nil, errInvalidArgument().Trace(event)
		}
	}

	// Set up a watchpoint listening for events within a directory tree rooted
	// at current working directory. Dispatch remove events to c.
	recursivePath := f.PathURL.Path
	if params.recursive {
		recursivePath = f.PathURL.Path + "..."
	}
	if e := notify.Watch(recursivePath, in, fsEvents...); e != nil {
		return nil, probe.NewError(e)
	}

	// wait for doneChan to close the watcher, eventChan and errorChan
	go func() {
		<-doneChan

		close(eventChan)
		close(errorChan)
		notify.Stop(in)
	}()

	timeFormatFS := "2006-01-02T15:04:05.000Z"

	// Get fsnotify notifications for events and errors, and sent them
	// using eventChan and errorChan
	go func() {
		for event := range out {
			if isIgnoredFile(event.Path()) {
				continue
			}
			var i os.FileInfo
			if IsPutEvent(event.Event()) {
				// Look for any writes, send a response to indicate a full copy.
				var e error
				i, e = os.Stat(event.Path())
				if e != nil {
					if os.IsNotExist(e) {
						continue
					}
					errorChan <- probe.NewError(e)
					continue
				}
				if i.IsDir() {
					// we want files
					continue
				}
				eventChan <- EventInfo{
					Time: UTCNow().Format(timeFormatFS),
					Size: i.Size(),
					Path: event.Path(),
					Type: EventCreate,
				}
			} else if IsDeleteEvent(event.Event()) {
				eventChan <- EventInfo{
					Time: UTCNow().Format(timeFormatFS),
					Path: event.Path(),
					Type: EventRemove,
				}
			} else if IsGetEvent(event.Event()) {
				eventChan <- EventInfo{
					Time: UTCNow().Format(timeFormatFS),
					Path: event.Path(),
					Type: EventAccessed,
				}
			}
		}
	}()

	return &watchObject{
		eventInfoChan: eventChan,
		errorChan:     errorChan,
		doneChan:      doneChan,
	}, nil
}
487201
func (t *Txn) GetWatch(k []byte) (<-chan struct{}, interface{}, bool) {
	return t.root.GetWatch(k)
}
603361
func (c *FakeAuditSinks) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(auditsinksResource, opts))
}
1006048
func (c *FakeNodes) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(nodesResource, opts))
}
944292
func (q *Queue) Watch() (eventq chan events.Event, cancel func()) {
	return q.CallbackWatch(nil)
}
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
	}
}
470564
func NewNotifyWatcher(caller base.APICaller, result params.NotifyWatchResult) watcher.NotifyWatcher {
	w := &notifyWatcher{
		caller:          caller,
		notifyWatcherId: result.NotifyWatcherId,
		out:             make(chan struct{}),
	}
	w.tomb.Go(w.loop)
	return w
}
494701
func (t *recursiveTree) 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
	}
	eventset := joinevents(events)
	if isrec {
		eventset |= recursive
	}
	t.rw.Lock()
	defer t.rw.Unlock()
	// case 1: cur is a child
	//
	// Look for parent watch which already covers the given path.
	parent := node{}
	self := false
	err = t.root.WalkPath(path, func(nd node, isbase bool) error {
		if watchTotal(nd) != 0 {
			parent = nd
			self = isbase
			return errSkip
		}
		return nil
	})
	cur := t.root.Add(path) // add after the walk, so it's less to traverse
	if err == nil && parent.Watch != nil {
		// Parent watch found. Register inactive watchpoint, so we have enough
		// information to shrink the eventset on eventual Stop.
		// return t.resetwatchpoint(parent, parent, c, eventset|inactive)
		var diff eventDiff
		if self {
			diff = watchAdd(cur, c, eventset)
		} else {
			diff = watchAddInactive(parent, c, eventset)
		}
		switch {
		case diff == none:
			// the parent watchpoint already covers requested subtree with its
			// eventset
		case diff[0] == 0:
			// TODO(rjeczalik): cleanup this panic after implementation is stable
			panic("dangling watchpoint: " + parent.Name)
		default:
			if isrec || watchIsRecursive(parent) {
				err = t.w.RecursiveRewatch(parent.Name, parent.Name, diff[0], diff[1])
			} else {
				err = t.w.Rewatch(parent.Name, diff[0], diff[1])
			}
			if err != nil {
				watchDel(parent, c, diff.Event())
				return err
			}
			watchAdd(cur, c, eventset)
			// TODO(rjeczalik): account top-most path for c
			return nil
		}
		if !self {
			watchAdd(cur, c, eventset)
		}
		return nil
	}
	// case 2: cur is new parent
	//
	// Look for children nodes, unwatch n-1 of them and rewatch the last one.
	var children []node
	fn := func(nd node) error {
		if len(nd.Watch) == 0 {
			return nil
		}
		children = append(children, nd)
		return errSkip
	}
	switch must(cur.Walk(fn)); len(children) {
	case 0:
		// no child watches, cur holds a new watch
	case 1:
		watchAdd(cur, c, eventset) // TODO(rjeczalik): update cache c subtree root?
		watchCopy(children[0], cur)
		err = t.w.RecursiveRewatch(children[0].Name, cur.Name, watchTotal(children[0]),
			watchTotal(cur))
		if err != nil {
			// Clean inactive watchpoint. The c chan did not exist before.
			cur.Child[""] = node{}
			delete(cur.Watch, c)
			return err
		}
		return nil
	default:
		watchAdd(cur, c, eventset)
		// Copy children inactive watchpoints to the new parent.
		for _, nd := range children {
			watchCopy(nd, cur)
		}
		// Watch parent subtree.
		if err = t.w.RecursiveWatch(cur.Name, watchTotal(cur)); err != nil {
			// Clean inactive watchpoint. The c chan did not exist before.
			cur.Child[""] = node{}
			delete(cur.Watch, c)
			return err
		}
		// Unwatch children subtrees.
		var e error
		for _, nd := range children {
			if watchIsRecursive(nd) {
				e = t.w.RecursiveUnwatch(nd.Name)
			} else {
				e = t.w.Unwatch(nd.Name)
			}
			if e != nil {
				err = nonil(err, e)
				// TODO(rjeczalik): child is still watched, warn all its watchpoints
				// about possible duplicate events via Error event
			}
		}
		return err
	}
	// case 3: cur is new, alone node
	switch diff := watchAdd(cur, c, eventset); {
	case diff == none:
		// TODO(rjeczalik): cleanup this panic after implementation is stable
		panic("watch requested but no parent watchpoint found: " + cur.Name)
	case diff[0] == 0:
		if isrec {
			err = t.w.RecursiveWatch(cur.Name, diff[1])
		} else {
			err = t.w.Watch(cur.Name, diff[1])
		}
		if err != nil {
			watchDel(cur, c, diff.Event())
			return err
		}
	default:
		// TODO(rjeczalik): cleanup this panic after implementation is stable
		panic("watch requested but no parent watchpoint found: " + cur.Name)
	}
	return nil
}
768747
func (fsh *Handler) Watch(paths []string, onEvent func(event fsnotify.Event), onClose func()) error {
	var err error
	fsh.watcher, err = fsnotify.NewWatcher()
	if err != nil {
		return errors.Errorf("failed to init fileDB file system watcher: %v", err)
	}
	for _, path := range paths {
		fsh.watcher.Add(path)
	}

	go func() {
		for {
			select {
			case event, ok := <-fsh.watcher.Events:
				if !ok {
					onClose()
					return
				}
				onEvent(event)
			case err := <-fsh.watcher.Errors:
				if err != nil {
					fsh.log.Errorf("filesystem notification error %v", err)
				}
			}
		}
	}()

	return nil
}
554650
func (t *recursiveTree) Stop(c chan<- EventInfo) {
	var err error
	fn := func(nd node) (e error) {
		diff := watchDel(nd, c, all)
		switch {
		case diff == none && watchTotal(nd) == 0:
			// TODO(rjeczalik): There's no watchpoints deeper in the tree,
			// probably we should remove the nodes as well.
			return nil
		case diff == none:
			// Removing c from nd does not require shrinking its eventset.
		case diff[1] == 0:
			if watchIsRecursive(nd) {
				e = t.w.RecursiveUnwatch(nd.Name)
			} else {
				e = t.w.Unwatch(nd.Name)
			}
		default:
			if watchIsRecursive(nd) {
				e = t.w.RecursiveRewatch(nd.Name, nd.Name, diff[0], diff[1])
			} else {
				e = t.w.Rewatch(nd.Name, diff[0], diff[1])
			}
		}
		fn := func(nd node) error {
			watchDel(nd, c, all)
			return nil
		}
		err = nonil(err, e, nd.Walk(fn))
		// TODO(rjeczalik): if e != nil store dummy chan in nd.Watch just to
		// retry un/rewatching next time and/or let the user handle the failure
		// vie Error event?
		return errSkip
	}
	t.rw.Lock()
	e := t.root.Walk("", fn) // TODO(rjeczalik): use max root per c
	t.rw.Unlock()
	if e != nil {
		err = nonil(err, e)
	}
	dbgprintf("Stop(%p) error: %v\n", c, err)
}
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
}
1104528
func nodesWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		catalog := p.client.Catalog()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		nodes, meta, err := catalog.Nodes(&opts)
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), nodes, err
	}
	return fn, nil
}
374982
func (m *Machine) WatchLXDProfileVerificationNeeded() (watcher.NotifyWatcher, error) {
	var results params.NotifyWatchResults
	args := params.Entities{
		Entities: []params.Entity{{Tag: m.tag.String()}},
	}
	err := m.facade.FacadeCall("WatchLXDProfileVerificationNeeded", args, &results)
	if err != nil {
		return nil, err
	}
	if len(results.Results) != 1 {
		return nil, fmt.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}
	return apiwatcher.NewNotifyWatcher(m.facade.RawAPICaller(), result), nil
}
897131
func (u *Unit) Watch() NotifyWatcher {
	return newEntityWatcher(u.st, unitsC, u.doc.DocID)
}
56038
func (a *Application) WatchLeaderSettings() NotifyWatcher {
	docId := a.st.docID(leadershipSettingsKey(a.Name()))
	return newEntityWatcher(a.st, settingsC, docId)
}
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
}