// Watch ranges over the passed in event chan and processes the events based on the
// handlers created for a given action.
// To stop watching, close the event chan.
519670
func (w *Handler) Watch(c <-chan eventtypes.Message) {
	for e := range c {
		w.mu.Lock()
		h, exists := w.handlers[w.keyFunc(e)]
		w.mu.Unlock()
		if !exists {
			continue
		}
		go h(e)
	}
}
193210
func (c *FakeMiddlewares) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(middlewaresResource, c.ns, opts))

}
1125513
func (c *FakeCustomResourceDefinitions) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(customresourcedefinitionsResource, opts))
}
159790
func (self *Bill) Watch(eventsChannel *EventChannel) error {
	self.stop = make(chan struct{})
	go func() {
		for {
			select {
			case event := <-eventsChannel.channel:
				switch {
				case event.EventAction == alerts.ONBOARD:
					err := self.OnboardFunc(event)
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				case event.EventAction == alerts.INSUFFICIENT_FUND:
					err := self.insufficientFund(event)
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				case event.EventAction == alerts.DEDUCT && self.M[constants.ENABLED] == constants.TRUE:
					err := self.deduct(event)
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				case event.EventAction == alerts.BILLEDHISTORY:
					err := self.billedhistory(event)
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				case event.EventAction == alerts.TRANSACTION:
					err := self.transaction(event)
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				case event.EventAction == alerts.SKEWS_ACTIONS || event.EventAction == alerts.QUOTA_UNPAID:
					err := self.skews(event)
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				}
			case <-self.stop:
				log.Info("bill watcher exiting")
				return
			}
		}
	}()
	return nil
}
1374122
func (pdb *BytesBrokerWatcherRedis) Watch(resp func(keyval.BytesWatchResp), closeChan chan string, keys ...string) error {
	if pdb.delegate.closed {
		return fmt.Errorf("watch(%v) called on a closed connection", keys)
	}
	return watch(pdb.delegate, resp, closeChan, pdb.addPrefix, pdb.trimPrefix, keys...)
}
1200735
func (mock *Mock) Watch(paths []string, onEvent func(event fsnotify.Event), onClose func()) error {
	go func() {
		for {
			select {
			case event, ok := <-mock.eventChan:
				if !ok {
					onClose()
					return
				}
				onEvent(event)
			}
		}
	}()
	return nil
}
763894
func (c *FakeEvents) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(eventsResource, c.ns, opts))

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

}
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
}
96114
func (api *API) Watch() (watcher.StringsWatcher, error) {
	var result params.StringsWatchResult
	err := api.caller.FacadeCall("Watch", nil, &result)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if result.Error != nil {
		return nil, errors.Trace(result.Error)
	}
	w := api.newWatcher(api.caller.RawAPICaller(), result)
	return w, nil
}
54560
func (c *Int) Watch(watch IntWatch) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.watches = append(c.watches, watch)
}
1071204
func (c *Client) Watch() (watcher.NotifyWatcher, error) {
	var result params.NotifyWatchResult
	err := c.caller.FacadeCall("Watch", nil, &result)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if result.Error != nil {
		return nil, result.Error
	}
	return c.newWatcher(c.caller.RawAPICaller(), result), nil
}
157597
func (r *Request) Watch() (watch.Watch, error) {
	if r.err != nil {
		return nil, r.err
	}

	r.params.Set("watch", "true")

	req, err := r.request()
	if err != nil {
		return nil, err
	}

	w, err := watch.NewBodyWatcher(req, r.client)
	return w, err
}
661134
func (c *FakeServices) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(servicesResource, c.ns, opts))

}
1281997
func (adapter *Adapter) Watch(resyncName string, changeChan chan datasync.ChangeEvent,
	resyncChan chan datasync.ResyncEvent, keyPrefixes ...string) (datasync.WatchRegistration, error) {

	if adapter.Watcher != nil {
		return adapter.Watcher.Watch(resyncName, changeChan, resyncChan, keyPrefixes...)
	}
	logrus.DefaultLogger().Debug("KeyValProtoWatcher is nil")

	return nil, nil
}
269057
func (f FakeWatcher) Watch(api.ListOptions) (watch.Interface, error) {
	return f.IFace, f.Err
}
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
}