// servicesWatch is used to watch the list of available services
763708
func servicesWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		catalog := p.client.Catalog()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		services, meta, err := catalog.Services(&opts)
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), services, err
	}
	return fn, nil
}
669437
func (c *FakePriorityClasses) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(priorityclassesResource, opts))
}
600036
func serveWatch(watcher watch.Interface, scope *RequestScope, mediaTypeOptions negotiation.MediaTypeOptions, req *http.Request, w http.ResponseWriter, timeout time.Duration) {
	options, err := optionsForTransform(mediaTypeOptions, req)
	if err != nil {
		scope.err(err, w, req)
		return
	}

	// negotiate for the stream serializer from the scope's serializer
	serializer, err := negotiation.NegotiateOutputMediaTypeStream(req, scope.Serializer, scope)
	if err != nil {
		scope.err(err, w, req)
		return
	}
	framer := serializer.StreamSerializer.Framer
	streamSerializer := serializer.StreamSerializer.Serializer
	encoder := scope.Serializer.EncoderForVersion(streamSerializer, scope.Kind.GroupVersion())
	useTextFraming := serializer.EncodesAsText
	if framer == nil {
		scope.err(fmt.Errorf("no framer defined for %q available for embedded encoding", serializer.MediaType), w, req)
		return
	}
	// TODO: next step, get back mediaTypeOptions from negotiate and return the exact value here
	mediaType := serializer.MediaType
	if mediaType != runtime.ContentTypeJSON {
		mediaType += ";stream=watch"
	}

	// locate the appropriate embedded encoder based on the transform
	var embeddedEncoder runtime.Encoder
	contentKind, contentSerializer, transform := targetEncodingForTransform(scope, mediaTypeOptions, req)
	if transform {
		info, ok := runtime.SerializerInfoForMediaType(contentSerializer.SupportedMediaTypes(), serializer.MediaType)
		if !ok {
			scope.err(fmt.Errorf("no encoder for %q exists in the requested target %#v", serializer.MediaType, contentSerializer), w, req)
			return
		}
		embeddedEncoder = contentSerializer.EncoderForVersion(info.Serializer, contentKind.GroupVersion())
	} else {
		embeddedEncoder = scope.Serializer.EncoderForVersion(serializer.Serializer, contentKind.GroupVersion())
	}

	ctx := req.Context()

	server := &WatchServer{
		Watching: watcher,
		Scope:    scope,

		UseTextFraming:  useTextFraming,
		MediaType:       mediaType,
		Framer:          framer,
		Encoder:         encoder,
		EmbeddedEncoder: embeddedEncoder,

		Fixup: func(obj runtime.Object) runtime.Object {
			result, err := transformObject(ctx, obj, options, mediaTypeOptions, scope, req)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("failed to transform object %v: %v", reflect.TypeOf(obj), err))
				return obj
			}
			// When we are transformed to a table, use the table options as the state for whether we
			// should print headers - on watch, we only want to print table headers on the first object
			// and omit them on subsequent events.
			if tableOptions, ok := options.(*metav1beta1.TableOptions); ok {
				tableOptions.NoHeaders = true
			}
			return result
		},

		TimeoutFactory: &realTimeoutFactory{timeout},
	}

	server.ServeHTTP(w, req)
}
8459
func (c *FakeRuntimeClasses) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(runtimeclassesResource, opts))
}
1165511
func (c *FakeResourceQuotas) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(resourcequotasResource, c.ns, opts))

}
759654
func (c *FakeDaemonSets) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(daemonsetsResource, c.ns, opts))

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

}
1046438
func (cp *ConsulClient) WatchService(srvName string, eventCh chan WatchServiceEvent, stopCh chan bool) error {
	keyName := "contiv.io/service/" + srvName + "/"

	// Run in background
	go func() {
		var currSrvMap = make(map[string]ServiceInfo)

		// Get current list of services
		srvList, lastIdx, err := cp.getServiceInstances(keyName, 0)
		if err != nil {
			log.Errorf("Error getting service instances for (%s): Err: %v", srvName, err)
		} else {
			// for each instance trigger an add event
			for _, srvInfo := range srvList {
				eventCh <- WatchServiceEvent{
					EventType:   WatchServiceEventAdd,
					ServiceInfo: srvInfo,
				}

				// Add the service to local cache
				srvKey := srvInfo.HostAddr + ":" + strconv.Itoa(srvInfo.Port)
				currSrvMap[srvKey] = srvInfo
			}
		}

		// Loop till asked to stop
		for {
			// Check if we should quit
			select {
			case <-stopCh:
				return
			default:
				// Read the service instances
				srvList, lastIdx, err = cp.getServiceInstances(keyName, lastIdx)
				if err != nil {
					if api.IsServerError(err) || strings.Contains(err.Error(), "EOF") || strings.Contains(err.Error(), "connection refused") {
						log.Warnf("Consul service watch: server error: %v Retrying..", err)
					} else {
						log.Errorf("Error getting service instances for (%s): Err: %v. Exiting watch", srvName, err)
					}

					// Wait a little and continue
					time.Sleep(5 * time.Second)
					continue
				} else {
					log.Debugf("Got consul srv list: {%+v}. Curr: {%+v}", srvList, currSrvMap)
					var newSrvMap = make(map[string]ServiceInfo)

					// Check if there are any new services
					for _, srvInfo := range srvList {
						srvKey := srvInfo.HostAddr + ":" + strconv.Itoa(srvInfo.Port)

						// If the entry didnt exists previously, trigger add event
						if _, ok := currSrvMap[srvKey]; !ok {
							log.Debugf("Sending add event for srv: %v", srvInfo)
							eventCh <- WatchServiceEvent{
								EventType:   WatchServiceEventAdd,
								ServiceInfo: srvInfo,
							}
						}

						// create new service map
						newSrvMap[srvKey] = srvInfo
					}

					// for all entries in old service list, see if we need to delete any
					for _, srvInfo := range currSrvMap {
						srvKey := srvInfo.HostAddr + ":" + strconv.Itoa(srvInfo.Port)

						// if the entry does not exists in new list, delete it
						if _, ok := newSrvMap[srvKey]; !ok {
							log.Debugf("Sending delete event for srv: %v", srvInfo)
							eventCh <- WatchServiceEvent{
								EventType:   WatchServiceEventDel,
								ServiceInfo: srvInfo,
							}
						}
					}

					// set new srv map as the current
					currSrvMap = newSrvMap
				}
			}
		}
	}()
	return nil
}
835383
func (c *FakeTemplates) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(templatesResource, c.ns, opts))

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

}
327010
func (s *store) WatchList(ctx context.Context, key string, resourceVersion string, pred storage.SelectionPredicate) (watch.Interface, error) {
	return s.watch(ctx, key, resourceVersion, pred, true)
}
92049
func (c *FakeDeployments) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(deploymentsResource, c.ns, opts))

}
904482
func (c *FakeRoleBindingRestrictions) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(rolebindingrestrictionsResource, c.ns, opts))

}
673439
func (s *podSecurityPolicy) Watch(opts api.ListOptions) (watch.Interface, error) {
	return s.client.Get().
		Prefix("watch").
		Resource("podsecuritypolicies").
		VersionedParams(&opts, api.ParameterCodec).
		Watch()
}
317158
func (a *AgentEntityWatcher) Watch(args params.Entities) (params.NotifyWatchResults, error) {
	result := params.NotifyWatchResults{
		Results: make([]params.NotifyWatchResult, len(args.Entities)),
	}
	if len(args.Entities) == 0 {
		return result, nil
	}
	canWatch, err := a.getCanWatch()
	if err != nil {
		return params.NotifyWatchResults{}, errors.Trace(err)
	}
	for i, entity := range args.Entities {
		tag, err := names.ParseTag(entity.Tag)
		if err != nil {
			result.Results[i].Error = ServerError(ErrPerm)
			continue
		}
		err = ErrPerm
		watcherId := ""
		if canWatch(tag) {
			watcherId, err = a.watchEntity(tag)
		}
		result.Results[i].NotifyWatcherId = watcherId
		result.Results[i].Error = ServerError(err)
	}
	return result, nil
}
897131
func (u *Unit) Watch() NotifyWatcher {
	return newEntityWatcher(u.st, unitsC, u.doc.DocID)
}