// Watch returns a watch.Interface that watches the requested events.
763894
func (c *FakeEvents) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(eventsResource, c.ns, opts))

}
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
}
759654
func (c *FakeDaemonSets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(daemonsetsResource, c.ns, opts))

}
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
}
909868
func (c *FakeControllerRevisions) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(controllerrevisionsResource, c.ns, opts))

}
1059213
func (c *FakeRoles) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(rolesResource, c.ns, opts))

}
626213
func (c *FakeStatefulSets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(statefulsetsResource, c.ns, opts))

}
406278
func (w *Watcher) watch(path string) error {
	w.ewmut.Lock()
	w.externalWatches[path] = true
	w.ewmut.Unlock()
	return w.addWatch(path, sys_NOTE_ALLEVENTS)
}
1083792
func (c *nodes) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Namespace(api.NamespaceAll).
		Resource(c.resourceName()).
		VersionedParams(&opts, api.Scheme).
		Watch()
}
738673
func (u *Unit) Watch() (watcher.NotifyWatcher, error) {
	return common.Watch(u.st.facade, "Watch", u.tag)
}
264985
func (self *User) Watch(eventsChannel *EventChannel) error {
	self.stop = make(chan struct{})
	go func() {
		for {
			select {
			case event := <-eventsChannel.channel:
				err := self.alert(event)
				if err != nil {
					log.Warningf("Failed to process watch event: %v", err)
				}
			case <-self.stop:
				log.Info("user watcher exiting")
				return
			}
		}
	}()
	return nil
}
78692
func (r *Request) Watch() (watch.Interface, error) {
	// We specifically don't want to rate limit watches, so we
	// don't use r.throttle here.
	if r.err != nil {
		return nil, r.err
	}
	url := r.URL().String()
	req, err := http.NewRequest(r.verb, url, r.body)
	if err != nil {
		return nil, err
	}
	client := r.client
	if client == nil {
		client = http.DefaultClient
	}
	time.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
	resp, err := client.Do(req)
	updateURLMetrics(r, resp, err)
	if r.baseURL != nil {
		if err != nil {
			r.backoffMgr.UpdateBackoff(r.baseURL, err, 0)
		} else {
			r.backoffMgr.UpdateBackoff(r.baseURL, err, resp.StatusCode)
		}
	}
	if err != nil {
		// The watch stream mechanism handles many common partial data errors, so closed
		// connections can be retried in many cases.
		if net.IsProbableEOF(err) {
			return watch.NewEmptyWatch(), nil
		}
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		defer resp.Body.Close()
		if result := r.transformResponse(resp, req); result.err != nil {
			return nil, result.err
		}
		return nil, fmt.Errorf("for request '%+v', got status: %v", url, resp.StatusCode)
	}
	return watch.NewStreamWatcher(watchjson.NewDecoder(resp.Body, r.content.Codec)), nil
}
719548
func (w *eventHandler) Watch(c <-chan eventtypes.Message) {
	for e := range c {
		w.mu.Lock()
		h, exists := w.handlers[e.Action]
		w.mu.Unlock()
		if !exists {
			continue
		}
		logrus.Debugf("event handler: received event: %v", e)
		go h(e)
	}
}
83374
func (f *fsClient) Watch(params watchParams) (*watchObject, *probe.Error) {
	eventChan := make(chan EventInfo)
	errorChan := make(chan *probe.Error)
	doneChan := make(chan bool)
	// Make the channel buffered to ensure no event is dropped. Notify will drop
	// an event if the receiver is not able to keep up the sending pace.
	in, out := PipeChan(1000)

	var fsEvents []notify.Event
	for _, event := range params.events {
		switch event {
		case "put":
			fsEvents = append(fsEvents, EventTypePut...)
		case "delete":
			fsEvents = append(fsEvents, EventTypeDelete...)
		case "get":
			fsEvents = append(fsEvents, EventTypeGet...)
		default:
			return nil, errInvalidArgument().Trace(event)
		}
	}

	// Set up a watchpoint listening for events within a directory tree rooted
	// at current working directory. Dispatch remove events to c.
	recursivePath := f.PathURL.Path
	if params.recursive {
		recursivePath = f.PathURL.Path + "..."
	}
	if e := notify.Watch(recursivePath, in, fsEvents...); e != nil {
		return nil, probe.NewError(e)
	}

	// wait for doneChan to close the watcher, eventChan and errorChan
	go func() {
		<-doneChan

		close(eventChan)
		close(errorChan)
		notify.Stop(in)
	}()

	timeFormatFS := "2006-01-02T15:04:05.000Z"

	// Get fsnotify notifications for events and errors, and sent them
	// using eventChan and errorChan
	go func() {
		for event := range out {
			if isIgnoredFile(event.Path()) {
				continue
			}
			var i os.FileInfo
			if IsPutEvent(event.Event()) {
				// Look for any writes, send a response to indicate a full copy.
				var e error
				i, e = os.Stat(event.Path())
				if e != nil {
					if os.IsNotExist(e) {
						continue
					}
					errorChan <- probe.NewError(e)
					continue
				}
				if i.IsDir() {
					// we want files
					continue
				}
				eventChan <- EventInfo{
					Time: UTCNow().Format(timeFormatFS),
					Size: i.Size(),
					Path: event.Path(),
					Type: EventCreate,
				}
			} else if IsDeleteEvent(event.Event()) {
				eventChan <- EventInfo{
					Time: UTCNow().Format(timeFormatFS),
					Path: event.Path(),
					Type: EventRemove,
				}
			} else if IsGetEvent(event.Event()) {
				eventChan <- EventInfo{
					Time: UTCNow().Format(timeFormatFS),
					Path: event.Path(),
					Type: EventAccessed,
				}
			}
		}
	}()

	return &watchObject{
		eventInfoChan: eventChan,
		errorChan:     errorChan,
		doneChan:      doneChan,
	}, nil
}
348490
func (c *FakeJobs) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(jobsResource, c.ns, opts))

}
62194
func (m *MockIngressInterface) Watch(arg0 v1.ListOptions) (watch.Interface, error) {
	ret := m.ctrl.Call(m, "Watch", arg0)
	ret0, _ := ret[0].(watch.Interface)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
516873
func (e *Store) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
	label := labels.Everything()
	if options != nil && options.LabelSelector != nil {
		label = options.LabelSelector
	}
	field := fields.Everything()
	if options != nil && options.FieldSelector != nil {
		field = options.FieldSelector
	}
	predicate := e.PredicateFunc(label, field)

	resourceVersion := ""
	if options != nil {
		resourceVersion = options.ResourceVersion
		predicate.AllowWatchBookmarks = options.AllowWatchBookmarks
	}
	return e.WatchPredicate(ctx, predicate, resourceVersion)
}