// GetWatch is used to lookup a specific key, returning
// the watch channel, value and if it was found
487201
func (t *Txn) GetWatch(k []byte) (<-chan struct{}, interface{}, bool) {
	return t.root.GetWatch(k)
}
913451
func (pdb *protoWatcher) Watch(resp func(datasync.ProtoWatchResp), closeChan chan string, keys ...string) error {
	err := pdb.watcher.Watch(func(msg keyval.BytesWatchResp) {
		resp(NewWatchResp(pdb.serializer, msg))
	}, closeChan, keys...)
	if err != nil {
		return err
	}
	return nil
}
370254
func (s *Discovery) Watch(stopCh <-chan struct{}) (<-chan discovery.Entries, <-chan error) {
	ch := make(chan discovery.Entries)
	errCh := make(chan error)
	ticker := time.NewTicker(s.heartbeat)

	go func() {
		defer close(errCh)
		defer close(ch)

		// Send the initial entries if available.
		var currentEntries discovery.Entries
		var err error

		s.mu.Lock()
		if len(s.values) > 0 {
			currentEntries, err = discovery.CreateEntries(s.values)
		}
		s.mu.Unlock()

		if err != nil {
			errCh <- err
		} else if currentEntries != nil {
			ch <- currentEntries
		}

		// Periodically send updates.
		for {
			select {
			case <-ticker.C:
				s.mu.Lock()
				newEntries, err := discovery.CreateEntries(s.values)
				s.mu.Unlock()
				if err != nil {
					errCh <- err
					continue
				}

				// Check if the file has really changed.
				if !newEntries.Equals(currentEntries) {
					ch <- newEntries
				}
				currentEntries = newEntries
			case <-stopCh:
				ticker.Stop()
				return
			}
		}
	}()

	return ch, errCh
}
1250727
func (c *Client) Watch(resp func(keyval.BytesWatchResp), closeChan chan string, keys ...string) error {
	consulLogger.Debug("Watch:", keys)
	for _, k := range keys {
		if err := c.watch(resp, closeChan, k); err != nil {
			return err
		}
	}
	return nil
}
324725
func (o *os) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
	var wo registry.WatchOptions
	for _, o := range opts {
		o(&wo)
	}

	req := o.opts.Client.NewRequest("go.micro.srv.discovery", "Registry.Watch", &proto2.WatchRequest{
		Service: wo.Service,
	})

	var gstream client.Streamer
	var grr error

	for _, addr := range o.next() {
		stream, err := o.opts.Client.StreamRemote(context.TODO(), addr, req)
		if err != nil {
			grr = err
			continue
		}
		gstream = stream
		break
	}

	if grr != nil {
		return nil, grr
	}

	// send empty watch request
	if err := gstream.Send(&proto2.WatchRequest{}); err != nil {
		return nil, err
	}

	wc := &watchClient{gstream}
	return &watcher{wc}, 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...)
}
1065484
func (wr *protoWatchResp) GetValue(msg proto.Message) error {
	return wr.serializer.Unmarshal(wr.BytesWatchResp.GetValue(), msg)
}
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
}
342319
func (lw *ListWatch) Watch(options metav1.ListOptions) (watch.Interface, error) {
	return lw.WatchFunc(options)
}
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...)
}
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
	}
}
1145258
func (w WatchSet) Watch(timeoutCh <-chan time.Time) bool {
	if w == nil {
		return false
	}

	// Create a context that gets cancelled when the timeout is triggered
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	go func() {
		select {
		case <-timeoutCh:
			cancel()
		case <-ctx.Done():
		}
	}()

	return w.WatchCtx(ctx) == context.Canceled
}
624680
func (db *BytesConnectionRedis) Watch(resp func(keyval.BytesWatchResp), closeChan chan string, keys ...string) error {
	if db.closed {
		return fmt.Errorf("watch(%v) called on a closed connection", keys)
	}
	db.closeCh = closeChan

	return watch(db, resp, db.closeCh, nil, nil, keys...)
}
39953
func (adapter *Registry) Watch(resyncName string, changeChan chan datasync.ChangeEvent,
	resyncChan chan datasync.ResyncEvent, keyPrefixes ...string) (datasync.WatchRegistration, error) {

	adapter.access.Lock()
	defer adapter.access.Unlock()

	if _, found := adapter.subscriptions[resyncName]; found {
		return nil, errors.New("Already watching " + resyncName)
	}

	adapter.subscriptions[resyncName] = &Subscription{
		ResyncName:  resyncName,
		ChangeChan:  changeChan,
		ResyncChan:  resyncChan,
		CloseChan:   make(chan string),
		KeyPrefixes: keyPrefixes,
	}

	return &WatchDataReg{resyncName, adapter}, nil
}
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)),
	}
}
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()
}
278342
func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string) error {
	r := f.NewBuilder().
		Unstructured().
		NamespaceParam(o.Namespace).DefaultNamespace().AllNamespaces(o.AllNamespaces).
		FilenameParam(o.ExplicitNamespace, &o.FilenameOptions).
		LabelSelectorParam(o.LabelSelector).
		FieldSelectorParam(o.FieldSelector).
		ExportParam(o.Export).
		RequestChunksOf(o.ChunkSize).
		ResourceTypeOrNameArgs(true, args...).
		SingleResourceType().
		Latest().
		TransformRequests(o.transformRequests).
		Do()
	if err := r.Err(); err != nil {
		return err
	}
	infos, err := r.Infos()
	if err != nil {
		return err
	}
	if multipleGVKsRequested(infos) {
		return i18n.Errorf("watch is only supported on individual resources and resource collections - more than 1 resource was found")
	}

	info := infos[0]
	mapping := info.ResourceMapping()
	printer, err := o.ToPrinter(mapping, o.AllNamespaces, false)
	if err != nil {
		return err
	}
	obj, err := r.Object()
	if err != nil {
		return err
	}

	// watching from resourceVersion 0, starts the watch at ~now and
	// will return an initial watch event.  Starting form ~now, rather
	// the rv of the object will insure that we start the watch from
	// inside the watch window, which the rv of the object might not be.
	rv := "0"
	isList := meta.IsListType(obj)
	if isList {
		// the resourceVersion of list objects is ~now but won't return
		// an initial watch event
		rv, err = meta.NewAccessor().ResourceVersion(obj)
		if err != nil {
			return err
		}
	}

	writer := utilprinters.GetNewTabWriter(o.Out)

	tableGK := metainternal.SchemeGroupVersion.WithKind("Table").GroupKind()

	// print the current object
	if !o.WatchOnly {
		var objsToPrint []runtime.Object

		if isList {
			objsToPrint, _ = meta.ExtractList(obj)
		} else {
			objsToPrint = append(objsToPrint, obj)
		}
		for _, objToPrint := range objsToPrint {
			if o.IsHumanReadablePrinter && objToPrint.GetObjectKind().GroupVersionKind().GroupKind() != tableGK {
				// printing anything other than tables always takes the internal version, but the watch event uses externals
				internalGV := mapping.GroupVersionKind.GroupKind().WithVersion(runtime.APIVersionInternal).GroupVersion()
				objToPrint = attemptToConvertToInternal(objToPrint, legacyscheme.Scheme, internalGV)
			}
			if err := printer.PrintObj(objToPrint, writer); err != nil {
				return fmt.Errorf("unable to output the provided object: %v", err)
			}
		}
		writer.Flush()
	}

	// print watched changes
	w, err := r.Watch(rv)
	if err != nil {
		return err
	}

	first := true
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	intr := interrupt.New(nil, cancel)
	intr.Run(func() error {
		_, err := watchtools.UntilWithoutRetry(ctx, w, func(e watch.Event) (bool, error) {
			if !isList && first {
				// drop the initial watch event in the single resource case
				first = false
				return false, nil
			}

			// printing always takes the internal version, but the watch event uses externals
			// TODO fix printing to use server-side or be version agnostic
			objToPrint := e.Object
			if o.IsHumanReadablePrinter && objToPrint.GetObjectKind().GroupVersionKind().GroupKind() != tableGK {
				internalGV := mapping.GroupVersionKind.GroupKind().WithVersion(runtime.APIVersionInternal).GroupVersion()
				objToPrint = attemptToConvertToInternal(e.Object, legacyscheme.Scheme, internalGV)
			}
			if err := printer.PrintObj(objToPrint, writer); err != nil {
				return false, err
			}
			writer.Flush()
			return false, nil
		})
		return err
	})
	return nil
}