// Watch function watch the keys having `prefix` prefix, it watch all the keys when prefix is '*'.
// it return a event channel. error wil be returned only when key is empty.
// the return channel have the newest data with init action in it.
2421
func (w *BaseWatcher) Watch(prefix string, ctx context.Context) (<-chan *Event, error) {
	switch prefix {
	case "":
		return nil, fmt.Errorf("empty key")
	case "*":
		return w.Sender.Watch("", ctx), nil
	default:
		return w.Sender.Watch(prefix, ctx), nil
	}
}
511388
func (c *namespaces) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Resource("namespaces").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
51544
func (self *Addons) 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 <-self.stop:
				log.Info("addons watcher exiting")
				return
			}
		}
	}()
	return nil
}
1242675
func (st *StatsConn) Watch(ctx context.Context, filePath string) (current *WatchData, changes <-chan *WatchData, cancel CancelFunc) {
	startTime := time.Now()
	statsKey := []string{"Watch", st.cell}
	defer topoStatsConnTimings.Record(statsKey, startTime)
	return st.conn.Watch(ctx, filePath)
}
969627
func (w *BaseWatcher) Get(prefix string) (map[string]interface{}, error) {
	switch prefix {
	case "":
		return nil, fmt.Errorf("empty key")
	case "*":
		return w.Sender.GetAll(), nil
	default:
		return w.Sender.Get(prefix), nil
	}
}
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()
}
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...)
}
138836
func (d *ConsulStateDriver) WatchAllState(baseKey string, sType core.State,
	unmarshal func([]byte, interface{}) error, rsps chan core.WatchState) error {
	baseKey = processKey(baseKey)
	byteRsps := make(chan [2][]byte, 1)
	recvErr := make(chan error, 1)

	go channelStateEvents(d, sType, unmarshal, byteRsps, rsps, recvErr)

	err := d.WatchAll(baseKey, byteRsps)
	if err != nil {
		return err
	}

	err = <-recvErr
	return err

}
215762
func (i *Info) Watch(resourceVersion string) (watch.Interface, error) {
	return NewHelper(i.Client, i.Mapping).WatchSingle(i.Namespace, i.Name, resourceVersion)
}
263130
func (self *Container) Watch(eventsChannel *EventChannel) error {
	self.stop = make(chan struct{})
	go func() {
		for {
			select {
			case event := <-eventsChannel.channel:
				switch {
				case event.EventAction == alerts.LAUNCHED:
					err := self.create()
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				case event.EventAction == alerts.DESTROYED:
					err := self.destroy()
					if err != nil {
						log.Warningf("Failed to process watch event: %v", err)
					}
				}
			case <-self.stop:
				log.Info("container watcher exiting")
				return
			}
		}
	}()
	return nil
}
145131
func (facade *Facade) Watch(uuid string) (watcher.NotifyWatcher, error) {
	results := params.NotifyWatchResults{}
	err := facade.call("Watch", uuid, &results)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if count := len(results.Results); count != 1 {
		return nil, countError(count)
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, errors.Trace(result.Error)
	}
	apiCaller := facade.caller.RawAPICaller()
	watcher := facade.newWatcher(apiCaller, result)
	return watcher, nil
}
985856
func (c *FakeKeysets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(keysetsResource, c.ns, opts))

}
1186647
func (c *client) WatchPrefix(prefix string, ch chan struct{}) {
	c.wctx, c.wcf = context.WithCancel(c.ctx)
	c.watcher = clientv3.NewWatcher(c.cli)

	wch := c.watcher.Watch(c.wctx, prefix, clientv3.WithPrefix(), clientv3.WithRev(0))
	ch <- struct{}{}
	for wr := range wch {
		if wr.Canceled {
			return
		}
		ch <- struct{}{}
	}
}
1281688
func (r *readdcw) watch(path string, event Event, recursive bool) error {
	if event&^(All|fileNotifyChangeAll) != 0 {
		return errors.New("notify: unknown event")
	}

	r.Lock()
	defer r.Unlock()

	if wd, ok := r.m[path]; ok {
		dbgprint("watch: already exists")
		wd.filter &^= stateUnwatch
		return nil
	}

	if err := r.lazyinit(); err != nil {
		return err
	}

	wd, err := newWatched(r.cph, uint32(event), recursive, path)
	if err != nil {
		return err
	}

	r.m[path] = wd
	dbgprint("watch: new watch added")

	return nil
}
394485
func (c *limitRanges) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Namespace(c.ns).
		Resource("limitRanges").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
617818
func (this *base) watch(path registry.Key, handler func(Event)) (node *Node, err error) {
	if this.zk == nil {
		return nil, ErrNotInitialized
	}

	// Get the node
	node, err = this.zk.Get(path.Path())
	switch {
	case err == nil:

		switch path.(type) {
		// Specical case.  Node exists but watching for create
		case registry.Create:
			this.before = nil
			return nil, ErrInvalidState

		case registry.Delete, registry.Change:
			this.stop, err = node.Watch(handler)
			if err != nil {
				return
			}
			this.before = node

		case registry.Members:
			this.stop, err = node.WatchChildren(handler)
			if err != nil {
				return
			}
			this.before = node

		}
		if this.timer != nil {
			this.timer.Reset(this.timeout)
		}

	case err == ErrNotExist:

		switch path.(type) {
		// Specical case.  Node does not exist but watching for delete
		case registry.Delete:
			return nil, ErrInvalidState

		case registry.Create, registry.Change:
			this.stop, err = this.zk.Watch(path.Path(), handler)
			if err != nil {
				return
			}

		case registry.Members:
			this.stop, err = this.zk.WatchChildren(path.Path(), handler)
			if err != nil {
				return
			}

		}
		if this.timer != nil {
			this.timer.Reset(this.timeout)
		}
	default:
		return
	}
	return
}
1185362
func (s *store) Watch(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate) (watch.Interface, error) {
	return s.watch(ctx, key, resourceVersion, pred, false)
}