// Watch for changes on a "key"
// It returns a channel that will receive changes or pass
// on errors. Upon creation, the current value will first
// be sent to the channel. Providing a non-nil stopCh can
// be used to stop watching.
421630
func (s *Etcd) Watch(key string, ctx context.Context, recursive bool, index uint64) (<-chan *store.Event, error) {
	opts := &etcd.WatcherOptions{Recursive: recursive, AfterIndex: index}
	watcher := s.client.Watcher(s.normalize(key), opts)

	// watchCh is sending back events to the caller
	watchCh := make(chan *store.Event)

	go func() {
		defer close(watchCh)
		for {
			result, err := watcher.Next(ctx)

			if err != nil {
				// Push error value through the channel.
				watchCh <- errorEvent(key, err)
				return
			}
			watchCh <- &store.Event{
				Action:        actionMap[result.Action],
				Key:           result.Node.Key,
				ModifiedIndex: result.Node.ModifiedIndex,
				Data: []*store.KVPair{
					&store.KVPair{
						Key:       result.Node.Key,
						Value:     []byte(result.Node.Value),
						LastIndex: result.Node.ModifiedIndex,
					},
				},
			}
		}
	}()

	return watchCh, 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)),
	}
}
673605
func (w *Watcher) Unwatch(key string, ch chan<- Change) {
	w.sendReq(reqUnwatch{key, ch})
}
153145
func (e *etcdClient) Watch(w *Watcher) {
	localCache := watcherCache{}
	listSignalSent := false

	scopedLog := e.getLogger().WithFields(logrus.Fields{
		fieldWatcher: w,
		fieldPrefix:  w.prefix,
	})
	<-e.Connected()

reList:
	for {
		e.limiter.Wait(ctx.TODO())
		res, err := e.client.Get(ctx.Background(), w.prefix, client.WithPrefix(),
			client.WithSerializable())
		if err != nil {
			scopedLog.WithError(Hint(err)).Warn("Unable to list keys before starting watcher")
			continue
		}

		nextRev := res.Header.Revision + 1
		scopedLog.Debugf("List response from etcd len=%d: %+v", res.Count, res)

		if res.Count > 0 {
			for _, key := range res.Kvs {
				t := EventTypeCreate
				if localCache.Exists(key.Key) {
					t = EventTypeModify
				}

				localCache.MarkInUse(key.Key)
				scopedLog.Debugf("Emitting list result as %v event for %s=%v", t, key.Key, key.Value)

				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Key:   string(key.Key),
					Value: key.Value,
					Typ:   t,
				}
				trackEventQueued(string(key.Key), t, queueStart.End(true).Total())
			}
		}

		// More keys to be read, call Get() again
		if res.More {
			continue
		}

		// Send out deletion events for all keys that were deleted
		// between our last known revision and the latest revision
		// received via Get
		localCache.RemoveDeleted(func(k string) {
			event := KeyValueEvent{
				Key: k,
				Typ: EventTypeDelete,
			}

			scopedLog.Debugf("Emitting EventTypeDelete event for %s", k)
			queueStart := spanstat.Start()
			w.Events <- event
			trackEventQueued(k, EventTypeDelete, queueStart.End(true).Total())
		})

		// Only send the list signal once
		if !listSignalSent {
			w.Events <- KeyValueEvent{Typ: EventTypeListDone}
			listSignalSent = true
		}

	recreateWatcher:
		scopedLog.WithField(fieldRev, nextRev).Debug("Starting to watch a prefix")

		e.limiter.Wait(ctx.TODO())
		etcdWatch := e.client.Watch(ctx.Background(), w.prefix,
			client.WithPrefix(), client.WithRev(nextRev))
		for {
			select {
			case <-w.stopWatch:
				close(w.Events)
				w.stopWait.Done()
				return

			case r, ok := <-etcdWatch:
				if !ok {
					time.Sleep(50 * time.Millisecond)
					goto recreateWatcher
				}

				scopedLog := scopedLog.WithField(fieldRev, r.Header.Revision)

				if err := r.Err(); err != nil {
					// We tried to watch on a compacted
					// revision that may no longer exist,
					// recreate the watcher and try to
					// watch on the next possible revision
					if err == v3rpcErrors.ErrCompacted {
						scopedLog.WithError(Hint(err)).Debug("Tried watching on compacted revision")
					}

					// mark all local keys in state for
					// deletion unless the upcoming GET
					// marks them alive
					localCache.MarkAllForDeletion()

					goto reList
				}

				nextRev = r.Header.Revision + 1
				scopedLog.Debugf("Received event from etcd: %+v", r)

				for _, ev := range r.Events {
					event := KeyValueEvent{
						Key:   string(ev.Kv.Key),
						Value: ev.Kv.Value,
					}

					switch {
					case ev.Type == client.EventTypeDelete:
						event.Typ = EventTypeDelete
						localCache.RemoveKey(ev.Kv.Key)
					case ev.IsCreate():
						event.Typ = EventTypeCreate
						localCache.MarkInUse(ev.Kv.Key)
					default:
						event.Typ = EventTypeModify
						localCache.MarkInUse(ev.Kv.Key)
					}

					scopedLog.Debugf("Emitting %v event for %s=%v", event.Typ, event.Key, event.Value)

					queueStart := spanstat.Start()
					w.Events <- event
					trackEventQueued(string(ev.Kv.Key), event.Typ, queueStart.End(true).Total())
				}
			}
		}
	}
}
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
}
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
}
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)
	}
}
1146783
func (mem *memNamedMapping) Watch(subscriber string, callback func(idxmap.NamedMappingGenericEvent)) error {
	mem.Debug("Watch ", subscriber)

	_, found := mem.subscribers.LoadOrStore(subscriber, callback)
	if found {
		return fmt.Errorf("Already registered channel per subscriber ")
	}
	return nil
}
648461
func (w *Watcher) observe() {
	for {
		select {
		case ev, ok := <-w.fsEvent:
			if !ok {
				// We get here only when w.fsEvent is stopped when closing the watcher
				w.watched = nil
				w.watchers = nil
				w.dirs = nil
				close(w.fsEvent)
				w.fsEvent = nil
				return
			}
			w.parseEv(ev)
		}
	}
}
672811
func watchChannels(consumer *client.Consumer, cfg *client.Config) {

	for {
		select {
		case notification, more := <-cfg.RecvNotificationChan:
			if more {
				handleNotifcation(consumer, notification)
			}
		case err, more := <-cfg.RecvErrorChan:
			if more {
				fmt.Printf("Message Recv Errored: %v\n", err)
			}
		case msg, more := <-cfg.RecvMessageChan:
			if more {
				messageCallback(consumer, msg, *commit)
			}
		case <-consumer.GetCloseChannel():
			return
		}
	}
}
229226
func (c *consulClient) Watch(w *Watcher) {
	// Last known state of all KVPairs matching the prefix
	localState := map[string]consulAPI.KVPair{}
	nextIndex := uint64(0)

	qo := consulAPI.QueryOptions{
		WaitTime: time.Second,
	}

	for {
		// Initialize sleep time to a millisecond as we don't
		// want to sleep in between successful watch cycles
		sleepTime := 1 * time.Millisecond

		qo.WaitIndex = nextIndex
		pairs, q, err := c.KV().List(w.prefix, &qo)
		if err != nil {
			sleepTime = 5 * time.Second
			Trace("List of Watch failed", err, logrus.Fields{fieldPrefix: w.prefix, fieldWatcher: w.name})
		}

		if q != nil {
			nextIndex = q.LastIndex
		}

		// timeout while watching for changes, re-schedule
		if qo.WaitIndex != 0 && (q == nil || q.LastIndex == qo.WaitIndex) {
			goto wait
		}

		for _, newPair := range pairs {
			oldPair, ok := localState[newPair.Key]

			// Keys reported for the first time must be new
			if !ok {
				if newPair.CreateIndex != newPair.ModifyIndex {
					log.Debugf("consul: Previously unknown key %s received with CreateIndex(%d) != ModifyIndex(%d)",
						newPair.Key, newPair.CreateIndex, newPair.ModifyIndex)
				}

				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Typ:   EventTypeCreate,
					Key:   newPair.Key,
					Value: newPair.Value,
				}
				trackEventQueued(newPair.Key, EventTypeCreate, queueStart.End(true).Total())
			} else if oldPair.ModifyIndex != newPair.ModifyIndex {
				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Typ:   EventTypeModify,
					Key:   newPair.Key,
					Value: newPair.Value,
				}
				trackEventQueued(newPair.Key, EventTypeModify, queueStart.End(true).Total())
			}

			// Everything left on localState will be assumed to
			// have been deleted, therefore remove all keys in
			// localState that still exist in the kvstore
			delete(localState, newPair.Key)
		}

		for k, deletedPair := range localState {
			queueStart := spanstat.Start()
			w.Events <- KeyValueEvent{
				Typ:   EventTypeDelete,
				Key:   deletedPair.Key,
				Value: deletedPair.Value,
			}
			trackEventQueued(deletedPair.Key, EventTypeDelete, queueStart.End(true).Total())
			delete(localState, k)
		}

		for _, newPair := range pairs {
			localState[newPair.Key] = *newPair

		}

		// Initial list operation has been completed, signal this
		if qo.WaitIndex == 0 {
			w.Events <- KeyValueEvent{Typ: EventTypeListDone}
		}

	wait:
		select {
		case <-time.After(sleepTime):
		case <-w.stopWatch:
			close(w.Events)
			w.stopWait.Done()
			return
		}
	}
}
768178
func (s *EtcdElection) Watch(ctx context.Context, startIndex uint64) chan bool {
	results := make(chan bool)
	var cancel atomic.Value
	var stop int32

	go func() {
		select {
		case <-ctx.Done():
			cancel.Load().(context.CancelFunc)()
			return
		}
	}()

	// Create our initial watcher
	watcher := s.api.Watcher(s.conf.Election, nil)

	s.wg.Loop(func() bool {
		// this ensures we exit properly if the watcher isn't connected to etcd
		if atomic.LoadInt32(&stop) == 1 {
			return false
		}

		ctx, c := context.WithTimeout(context.Background(), time.Second*10)
		cancel.Store(c)
		resp, err := watcher.Next(ctx)
		if err != nil {
			if err == context.Canceled {
				close(results)
				return false
			}

			if cErr, ok := err.(etcd.Error); ok {
				if cErr.Code == etcd.ErrorCodeEventIndexCleared {
					logrus.WithField("category", "election").
						Infof("EtcdElection %s - new index is %d",
							err, cErr.Index+1)

					// Re-create the watcher with a newer index until we catch up
					watcher = s.api.Watcher(s.conf.Election, nil)
					return true
				}
			}
			if err != context.DeadlineExceeded {
				logrus.WithField("category", "election").
					Errorf("EtcdElection etcd error: %s", err)
				time.Sleep(time.Second * 10)
			}
		} else {
			logrus.WithField("category", "election").
				Debugf("EtcdElection etcd event: %+v\n", resp)
			results <- true
		}
		return true
	})
	return results
}
198458
func checksWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	var service, state string
	if err := assignValue(params, "service", &service); err != nil {
		return nil, err
	}
	if err := assignValue(params, "state", &state); err != nil {
		return nil, err
	}
	if service != "" && state != "" {
		return nil, fmt.Errorf("Cannot specify service and state")
	}
	if service == "" && state == "" {
		state = "any"
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		health := p.client.Health()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		var checks []*consulapi.HealthCheck
		var meta *consulapi.QueryMeta
		var err error
		if state != "" {
			checks, meta, err = health.State(state, &opts)
		} else {
			checks, meta, err = health.Checks(service, &opts)
		}
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), checks, err
	}
	return fn, nil
}
439383
func keyPrefixWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	var prefix string
	if err := assignValue(params, "prefix", &prefix); err != nil {
		return nil, err
	}
	if prefix == "" {
		return nil, fmt.Errorf("Must specify a single prefix to watch")
	}
	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		kv := p.client.KV()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		pairs, meta, err := kv.List(prefix, &opts)
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), pairs, err
	}
	return fn, nil
}
1166635
func (c *s3Client) Watch(params watchParams) (*watchObject, *probe.Error) {
	eventChan := make(chan EventInfo)
	errorChan := make(chan *probe.Error)
	doneChan := make(chan bool)

	// Extract bucket and object.
	bucket, object := c.url2BucketAndObject()

	// Flag set to set the notification.
	var events []string
	for _, event := range params.events {
		switch event {
		case "put":
			events = append(events, string(minio.ObjectCreatedAll))
		case "delete":
			events = append(events, string(minio.ObjectRemovedAll))
		case "get":
			events = append(events, string(minio.ObjectAccessedAll))
		default:
			return nil, errInvalidArgument().Trace(event)
		}
	}
	if object != "" && params.prefix != "" {
		return nil, errInvalidArgument().Trace(params.prefix, object)
	}
	if object != "" && params.prefix == "" {
		params.prefix = object
	}

	doneCh := make(chan struct{})

	// wait for doneChan to close the other channels
	go func() {
		<-doneChan

		close(doneCh)
		close(eventChan)
		close(errorChan)
	}()

	// Start listening on all bucket events.
	eventsCh := c.api.ListenBucketNotification(bucket, params.prefix, params.suffix, events, doneCh)

	wo := &watchObject{
		eventInfoChan: eventChan,
		errorChan:     errorChan,
		doneChan:      doneChan,
	}

	// wait for events to occur and sent them through the eventChan and errorChan
	go func() {
		defer wo.Close()
		for notificationInfo := range eventsCh {
			if notificationInfo.Err != nil {
				if nErr, ok := notificationInfo.Err.(minio.ErrorResponse); ok && nErr.Code == "APINotSupported" {
					errorChan <- probe.NewError(APINotImplemented{
						API:     "Watch",
						APIType: c.targetURL.Scheme + "://" + c.targetURL.Host,
					})
					return
				}
				errorChan <- probe.NewError(notificationInfo.Err)
			}

			for _, record := range notificationInfo.Records {
				bucketName := record.S3.Bucket.Name
				key, e := url.QueryUnescape(record.S3.Object.Key)
				if e != nil {
					errorChan <- probe.NewError(e)
					continue
				}

				u := *c.targetURL
				u.Path = path.Join(string(u.Separator), bucketName, key)
				if strings.HasPrefix(record.EventName, "s3:ObjectCreated:") {
					eventChan <- EventInfo{
						Time:      record.EventTime,
						Size:      record.S3.Object.Size,
						Path:      u.String(),
						Type:      EventCreate,
						Host:      record.Source.Host,
						Port:      record.Source.Port,
						UserAgent: record.Source.UserAgent,
					}

				} else if strings.HasPrefix(record.EventName, "s3:ObjectRemoved:") {
					eventChan <- EventInfo{
						Time:      record.EventTime,
						Path:      u.String(),
						Type:      EventRemove,
						Host:      record.Source.Host,
						Port:      record.Source.Port,
						UserAgent: record.Source.UserAgent,
					}
				} else if record.EventName == minio.ObjectAccessedGet {
					eventChan <- EventInfo{
						Time:      record.EventTime,
						Size:      record.S3.Object.Size,
						Path:      u.String(),
						Type:      EventAccessedRead,
						Host:      record.Source.Host,
						Port:      record.Source.Port,
						UserAgent: record.Source.UserAgent,
					}
				} else if record.EventName == minio.ObjectAccessedHead {
					eventChan <- EventInfo{
						Time:      record.EventTime,
						Size:      record.S3.Object.Size,
						Path:      u.String(),
						Type:      EventAccessedStat,
						Host:      record.Source.Host,
						Port:      record.Source.Port,
						UserAgent: record.Source.UserAgent,
					}
				}
			}
		}
	}()

	return wo, nil
}
792594
func (m *MockBaseWatcher) Watch(arg0 string, arg1 interface{}, arg2 chan<- watcher.Change) {
	m.ctrl.Call(m, "Watch", arg0, arg1, arg2)
}
985856
func (c *FakeKeysets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(keysetsResource, c.ns, opts))

}