// Watch starts new filesystem notification watcher. All events from files are passed to 'onEvent' function.
// Function 'onClose' is called when event channel is closed.
768747
func (fsh *Handler) Watch(paths []string, onEvent func(event fsnotify.Event), onClose func()) error {
	var err error
	fsh.watcher, err = fsnotify.NewWatcher()
	if err != nil {
		return errors.Errorf("failed to init fileDB file system watcher: %v", err)
	}
	for _, path := range paths {
		fsh.watcher.Add(path)
	}

	go func() {
		for {
			select {
			case event, ok := <-fsh.watcher.Events:
				if !ok {
					onClose()
					return
				}
				onEvent(event)
			case err := <-fsh.watcher.Errors:
				if err != nil {
					fsh.log.Errorf("filesystem notification error %v", err)
				}
			}
		}
	}()

	return nil
}
479628
func (c *FakeStorageClasses) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(storageclassesResource, opts))
}
256017
func (c *Client) Watch(resp func(response keyval.BytesWatchResp), closeChan chan string, keys ...string) error {
	c.Lock()
	defer c.Unlock()

	for _, key := range keys {
		dc := make(chan keyedData)
		go c.watch(resp, dc, closeChan, key)
		c.watchers[key] = dc
	}

	return nil
}
897131
func (u *Unit) Watch() NotifyWatcher {
	return newEntityWatcher(u.st, unitsC, u.doc.DocID)
}
438569
func (p *Project) Watch(wg *sync.WaitGroup) {
	var err error
	// change channel
	p.stop = make(chan bool)
	// init a new watcher
	p.watcher, err = NewFileWatcher(p.parent.Settings.Legacy)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		close(p.stop)
		p.watcher.Close()
	}()
	// before start checks
	p.Before()
	// start watcher
	go p.Reload("", p.stop)
L:
	for {
		select {
		case event := <-p.watcher.Events():
			if p.parent.Settings.Recovery.Events {
				log.Println("File:", event.Name, "LastFile:", p.last.file, "Time:", time.Now(), "LastTime:", p.last.time)
			}
			if time.Now().Truncate(time.Second).After(p.last.time) {
				// switch event type
				switch event.Op {
				case fsnotify.Chmod:
				case fsnotify.Remove:
					p.watcher.Remove(event.Name)
					if p.Validate(event.Name, false) && ext(event.Name) != "" {
						// stop and restart
						close(p.stop)
						p.stop = make(chan bool)
						p.Change(event)
						go p.Reload("", p.stop)
					}
				default:
					if p.Validate(event.Name, true) {
						fi, err := os.Stat(event.Name)
						if err != nil {
							continue
						}
						if fi.IsDir() {
							filepath.Walk(event.Name, p.walk)
						} else {
							// stop and restart
							close(p.stop)
							p.stop = make(chan bool)
							p.Change(event)
							go p.Reload(event.Name, p.stop)
							p.last.time = time.Now().Truncate(time.Second)
							p.last.file = event.Name
						}
					}
				}
			}
		case err := <-p.watcher.Errors():
			p.Err(err)
		case <-p.exit:
			p.After()
			break L
		}
	}
	wg.Done()
}
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
}
763894
func (c *FakeEvents) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(eventsResource, c.ns, opts))

}
648214
func (m *Machine) Watch() NotifyWatcher {
	return newEntityWatcher(m.st, machinesC, m.doc.DocID)
}
832446
func (s *Server) Watch(ctx context.Context, filePath string) (*topo.WatchData, <-chan *topo.WatchData, topo.CancelFunc) {
	nodePath := path.Join(s.root, filePath)

	// Get the initial version of the file
	initial, err := s.cli.Get(ctx, nodePath)
	if err != nil {
		// Generic error.
		return &topo.WatchData{Err: convertError(err, nodePath)}, nil, nil
	}
	if len(initial.Kvs) != 1 {
		// Node doesn't exist.
		return &topo.WatchData{Err: topo.NewError(topo.NoNode, nodePath)}, nil, nil
	}
	wd := &topo.WatchData{
		Contents: initial.Kvs[0].Value,
		Version:  EtcdVersion(initial.Kvs[0].ModRevision),
	}

	// Create a context, will be used to cancel the watch.
	watchCtx, watchCancel := context.WithCancel(context.Background())

	// Create the Watcher.  We start watching from the response we
	// got, not from the file original version, as the server may
	// not have that much history.
	watcher := s.cli.Watch(watchCtx, nodePath, clientv3.WithRev(initial.Header.Revision))
	if watcher == nil {
		return &topo.WatchData{Err: vterrors.Errorf(vtrpc.Code_INVALID_ARGUMENT, "Watch failed")}, nil, nil
	}

	// Create the notifications channel, send updates to it.
	notifications := make(chan *topo.WatchData, 10)
	go func() {
		defer close(notifications)

		var currVersion = initial.Header.Revision
		var watchRetries int
		for {
			select {

			case <-watchCtx.Done():
				// This includes context cancelation errors.
				notifications <- &topo.WatchData{
					Err: convertError(watchCtx.Err(), nodePath),
				}
				return
			case wresp, ok := <-watcher:
				if !ok {
					if watchRetries > 10 {
						time.Sleep(time.Duration(watchRetries) * time.Second)
					}
					watchRetries++
					newWatcher := s.cli.Watch(watchCtx, nodePath, clientv3.WithRev(currVersion))
					if newWatcher == nil {
						log.Warningf("watch %v failed and get a nil channel returned, currVersion: %v", nodePath, currVersion)
					} else {
						watcher = newWatcher
					}
					continue
				}

				watchRetries = 0

				if wresp.Canceled {
					// Final notification.
					notifications <- &topo.WatchData{
						Err: convertError(wresp.Err(), nodePath),
					}
					return
				}

				currVersion = wresp.Header.GetRevision()

				for _, ev := range wresp.Events {
					switch ev.Type {
					case mvccpb.PUT:
						notifications <- &topo.WatchData{
							Contents: ev.Kv.Value,
							Version:  EtcdVersion(ev.Kv.Version),
						}
					case mvccpb.DELETE:
						// Node is gone, send a final notice.
						notifications <- &topo.WatchData{
							Err: topo.NewError(topo.NoNode, nodePath),
						}
						return
					default:
						notifications <- &topo.WatchData{
							Err: vterrors.Errorf(vtrpc.Code_INTERNAL, "unexpected event received: %v", ev),
						}
						return
					}
				}
			}
		}
	}()

	return wd, notifications, topo.CancelFunc(watchCancel)
}
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
}
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
}
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
	}
}
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
}
955353
func (c *FakeClusterRoleBindings) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(clusterrolebindingsResource, opts))
}
21475
func (c *FakeClusterRoles) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(clusterrolesResource, opts))
}
1187855
func (sw *StackEventWatcher) Watch() error {
	if sw.seenEvents == nil {
		sw.seenEvents = map[string]struct{}{}
	}
	lastStackStatus := ""
	for {
		// print the events for the stack
		sw.Service.DescribeStackEventsPages(&cloudformation.DescribeStackEventsInput{
			StackName: aws.String(sw.StackName),
		}, func(p *cloudformation.DescribeStackEventsOutput, _ bool) bool {
			for _, stackEvent := range p.StackEvents {
				if _, ok := sw.seenEvents[*stackEvent.EventId]; ok {
					continue
				}
				wrapStrPtr := func(s *string) string {
					if s == nil {
						return ""
					}
					return *s
				}

				l := log.WithField("Status", *stackEvent.ResourceStatus)
				if stackEvent.ResourceType != nil {
					l = l.WithField("Type", *stackEvent.ResourceType)
				}
				if stackEvent.ResourceType != nil {
					l = l.WithField("Type", *stackEvent.ResourceType)
				}
				if stackEvent.PhysicalResourceId != nil {
					l = l.WithField("PhysicalID", *stackEvent.PhysicalResourceId)
				}
				if stackEvent.LogicalResourceId != nil {
					l = l.WithField("LogicalID", *stackEvent.LogicalResourceId)
				}
				if strings.Contains(*stackEvent.ResourceStatus, "FAIL") {
					l.Error(wrapStrPtr(stackEvent.ResourceStatusReason))
				} else {
					l.Info(wrapStrPtr(stackEvent.ResourceStatusReason))
				}

				sw.seenEvents[*stackEvent.EventId] = struct{}{}
			}
			return true
		})

		// monitor the status of the stack
		describeStacksResponse, err := sw.Service.DescribeStacks(&cloudformation.DescribeStacksInput{
			StackName: aws.String(sw.StackName),
		})
		if err != nil {
			// the stack might not exist yet
			log.Errorf("DescribeStacks: %s", err)
			time.Sleep(time.Second)
			continue
		}

		stackStatus := *describeStacksResponse.Stacks[0].StackStatus
		if stackStatus != lastStackStatus {
			log.Infof("Stack: %s\n", stackStatus)
			lastStackStatus = stackStatus
		}
		switch stackStatus {
		case cloudformation.StackStatusCreateComplete:
			return nil
		case cloudformation.StackStatusCreateFailed:
			return fmt.Errorf("%s", stackStatus)
		case cloudformation.StackStatusRollbackComplete:
			return fmt.Errorf("%s", stackStatus)
		case cloudformation.StackStatusUpdateRollbackComplete:
			return fmt.Errorf("%s", stackStatus)
		case cloudformation.StackStatusRollbackFailed:
			return fmt.Errorf("%s", stackStatus)
		case cloudformation.StackStatusUpdateComplete:
			return nil
		case cloudformation.StackStatusUpdateRollbackFailed:
			return fmt.Errorf("%s", stackStatus)
		default:
			time.Sleep(time.Second * 5)
			continue
		}
	}
}
843072
func (w *Watcher) watch(path string) error {
	return w.addWatch(path, sys_AGNOSTIC_EVENTS)
}