// startWatching does:
// - get current objects if initialRev=0; set initialRev to current rev
// - watch on given key and send events to process.
971410
func (wc *watchChan) startWatching(watchClosedCh chan struct{}) {
	if wc.initialRev == 0 {
		if err := wc.sync(); err != nil {
			klog.Errorf("failed to sync with latest state: %v", err)
			wc.sendError(err)
			return
		}
	}
	opts := []clientv3.OpOption{clientv3.WithRev(wc.initialRev + 1), clientv3.WithPrevKV()}
	if wc.recursive {
		opts = append(opts, clientv3.WithPrefix())
	}
	wch := wc.watcher.client.Watch(wc.ctx, wc.key, opts...)
	for wres := range wch {
		if wres.Err() != nil {
			err := wres.Err()
			// If there is an error on server (e.g. compaction), the channel will return it before closed.
			klog.Errorf("watch chan error: %v", err)
			wc.sendError(err)
			return
		}
		for _, e := range wres.Events {
			parsedEvent, err := parseEvent(e)
			if err != nil {
				klog.Errorf("watch chan error: %v", err)
				wc.sendError(err)
				return
			}
			wc.sendEvent(parsedEvent)
		}
	}
	// When we come to this point, it's only possible that client side ends the watch.
	// e.g. cancel the context, close the client.
	// If this watch chan is broken and context isn't cancelled, other goroutines will still hang.
	// We should notify the main thread that this goroutine has exited.
	close(watchClosedCh)
}
854008
func (c *FakeInstanceGroups) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(instancegroupsResource, c.ns, opts))

}
816939
func eventWatch(params map[string]interface{}) (WatcherFunc, error) {
	// The stale setting doesn't apply to events.

	var name string
	if err := assignValue(params, "name", &name); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		event := p.client.Event()
		opts := makeQueryOptionsWithContext(p, false)
		defer p.cancelFunc()
		events, meta, err := event.List(name, &opts)
		if err != nil {
			return nil, nil, err
		}

		// Prune to only the new events
		for i := 0; i < len(events); i++ {
			if WaitIndexVal(event.IDToIndex(events[i].ID)).Equal(p.lastParamVal) {
				events = events[i+1:]
				break
			}
		}
		return WaitIndexVal(meta.LastIndex), events, err
	}
	return fn, 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
}
383357
func (w *watcher) start() {
	if w.starting || w.running {
		return
	}
	w.starting = true
	go w.loop()
}
98118
func (st *state) WatchForRebootEvent() (watcher.NotifyWatcher, error) {
	var result params.NotifyWatchResult

	if err := st.facade.FacadeCall("WatchForRebootEvent", nil, &result); err != nil {
		return nil, err
	}
	if result.Error != nil {
		return nil, result.Error
	}

	w := apiwatcher.NewNotifyWatcher(st.facade.RawAPICaller(), result)
	return w, nil
}
22013
func (s *Sender) Watch(key string, ctx context.Context) <-chan *Event {
	s.Lock()
	defer s.Unlock()

	log.Infof("A new receiver watching to %s", key)
	r := &Receiver{
		key:     key,
		ctx:     ctx,
		ch:      make(chan *Event, 1),
		counter: 2,
	}
	s.receivers = append(s.receivers, r)
	return (<-chan *Event)(r.ch)
}
891121
func (c *consumer) Start() error {
	// Start consuming events from firehose.
	eventsCh, errCh := c.rawConsumer.Consume()

	// Construct default slowDetector
	sd := &defaultSlowDetector{
		logger: c.logger,
	}

	// Store slowDetector (for Close() fucntion)
	c.slowDetector = sd

	// Start reading events from firehose and detect `slowConsumerAlert`.
	// The detection is notified by detectCh.
	c.eventCh, c.errCh, c.detectCh = sd.Detect(eventsCh, errCh)

	// In current implementation no errors are happened.
	//
	// This is for preventing interfance change in future when
	// we need to put some other function here and it returns error.
	return nil
}
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
		}
	}
}
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
}
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
	}
}
1223635
func (w *identityWatcher) watch(owner IdentityAllocatorOwner, events allocator.AllocatorEventChan) {
	w.stopChan = make(chan bool)

	go func() {
		for {
			added := IdentityCache{}
			deleted := IdentityCache{}

		First:
			for {
				// Wait for one identity add or delete or stop
				select {
				case event, ok := <-events:
					if !ok {
						// 'events' was closed
						return
					}
					// Collect first added and deleted labels
					switch event.Typ {
					case kvstore.EventTypeCreate, kvstore.EventTypeDelete:
						if collectEvent(event, added, deleted) {
							// First event collected
							break First
						}
					default:
						// Ignore modify events
					}
				case <-w.stopChan:
					return
				}
			}

		More:
			for {
				// see if there is more, but do not wait nor stop
				select {
				case event, ok := <-events:
					if !ok {
						// 'events' was closed
						break More
					}
					// Collect more added and deleted labels
					switch event.Typ {
					case kvstore.EventTypeCreate, kvstore.EventTypeDelete:
						collectEvent(event, added, deleted)
					default:
						// Ignore modify events
					}
				default:
					// No more events available without blocking
					break More
				}
			}
			// Issue collected updates
			owner.UpdateIdentities(added, deleted)
		}
	}()
}
696358
func (o *Observer) Start() {
	for {
		select {
		case evt := <-o.notify: // We have an event to observe
			if len(o.observers) == 0 {
				log.Printf("Ignoring event due to no observers: %s", evt.ID)
				continue
			}

			evtb, err := json.Marshal(evt)
			if err != nil {
				log.Printf("Error marshaling event: %s", err)
				continue
			}

			sse := []byte("event: " + evt.Type + "\ndata: ")
			sse = append(sse, append(evtb, []byte("\n\n")...)...)

			for n := range o.observers {
				n <- sse
			}

		case n := <-o.newObservers:
			o.observers[n] = true
		case n := <-o.closedObservers:
			delete(o.observers, n)

		case <-o.closed: // The Observer has been closed.
			return
		}
	}
}
626213
func (c *FakeStatefulSets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(statefulsetsResource, c.ns, opts))

}
136949
func (m *Memory) NewWatcher(ctx context.Context, watch backend.Watch) (backend.Watcher, error) {
	if m.EventsOff {
		return nil, trace.BadParameter("events are turned off for this backend")
	}
	return m.buf.NewWatcher(ctx, watch)
}
1089282
func (s *Stopwatch) Start() {
	s.Lock()
	defer s.Unlock()

	if !s.isRunning() {
		s.refTime = time.Now()
	}
}
952901
func (s *GRPCServer) OnStart() error {
	if err := s.BaseService.OnStart(); err != nil {
		return err
	}
	ln, err := net.Listen(s.proto, s.addr)
	if err != nil {
		return err
	}
	s.Logger.Info("Listening", "proto", s.proto, "addr", s.addr)
	s.listener = ln
	s.server = grpc.NewServer()
	types.RegisterABCIApplicationServer(s.server, s.app)
	go s.server.Serve(s.listener)
	return nil
}