// Implements cache.ListerWatcher interface.
324419
func (lw *cacherListerWatcher) List(options metav1.ListOptions) (runtime.Object, error) {
	list := lw.newListFunc()
	pred := storage.SelectionPredicate{
		Label:    labels.Everything(),
		Field:    fields.Everything(),
		Limit:    options.Limit,
		Continue: options.Continue,
	}

	if err := lw.storage.List(context.TODO(), lw.resourcePrefix, "", pred, list); err != nil {
		return nil, err
	}
	return list, nil
}
710944
func (c *Client) WatchUnits(application string) (watcher.StringsWatcher, error) {
	applicationTag, err := applicationTag(application)
	if err != nil {
		return nil, errors.Trace(err)
	}
	args := entities(applicationTag)

	var results params.StringsWatchResults
	if err := c.facade.FacadeCall("WatchUnits", args, &results); err != nil {
		return nil, err
	}
	if n := len(results.Results); n != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", n)
	}
	if err := results.Results[0].Error; err != nil {
		return nil, errors.Trace(err)
	}
	w := apiwatcher.NewStringsWatcher(c.facade.RawAPICaller(), results.Results[0])
	return w, nil
}
979441
func (u *UnitsWatcher) WatchUnits(args params.Entities) (params.StringsWatchResults, error) {
	result := params.StringsWatchResults{
		Results: make([]params.StringsWatchResult, len(args.Entities)),
	}
	if len(args.Entities) == 0 {
		return result, nil
	}
	canWatch, err := u.getCanWatch()
	if err != nil {
		return params.StringsWatchResults{}, 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
		}
		entityResult, err := u.watchOneEntityUnits(canWatch, tag)
		result.Results[i] = entityResult
		result.Results[i].Error = ServerError(err)
	}
	return result, nil
}
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)
}
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()
}
289611
func (c *Client) Watch(application string) (watcher.NotifyWatcher, error) {
	tag, err := c.appTag(application)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return common.Watch(c.facade, "Watch", tag)
}
927610
func (sb *storageBackend) WatchStorageAttachment(s names.StorageTag, u names.UnitTag) NotifyWatcher {
	id := storageAttachmentId(u.Id(), s.Id())
	return newEntityWatcher(sb.mb, storageAttachmentsC, sb.mb.docID(id))
}
660526
func (c *Client) WatchRelationUnits(remoteRelationArg params.RemoteEntityArg) (watcher.RelationUnitsWatcher, error) {
	args := params.RemoteEntityArgs{Args: []params.RemoteEntityArg{remoteRelationArg}}
	// Use any previously cached discharge macaroons.
	if ms, ok := c.getCachedMacaroon("watch relation units", remoteRelationArg.Token); ok {
		args.Args[0].Macaroons = ms
	}

	var results params.RelationUnitsWatchResults
	apiCall := func() error {
		// Reset the results struct before each api call.
		results = params.RelationUnitsWatchResults{}
		if err := c.facade.FacadeCall("WatchRelationUnits", args, &results); err != nil {
			return errors.Trace(err)
		}
		if len(results.Results) != 1 {
			return errors.Errorf("expected 1 result, got %d", len(results.Results))
		}
		return nil
	}

	// Make the api call the first time.
	if err := apiCall(); err != nil {
		return nil, errors.Trace(err)
	}

	// On error, possibly discharge the macaroon and retry.
	result := results.Results[0]
	if result.Error != nil {
		mac, err := c.handleError(result.Error)
		if err != nil {
			result.Error.Message = err.Error()
			return nil, result.Error
		}
		args.Args[0].Macaroons = mac
		c.cache.Upsert(args.Args[0].Token, mac)

		if err := apiCall(); err != nil {
			return nil, errors.Trace(err)
		}
		result = results.Results[0]
	}
	if result.Error != nil {
		return nil, result.Error
	}

	w := apiwatcher.NewRelationUnitsWatcher(c.facade.RawAPICaller(), result)
	return w, nil
}
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
}
337380
func (sa *StorageAccessor) WatchUnitStorageAttachments(unitTag names.UnitTag) (watcher.StringsWatcher, error) {
	var results params.StringsWatchResults
	args := params.Entities{
		Entities: []params.Entity{{Tag: unitTag.String()}},
	}
	err := sa.facade.FacadeCall("WatchUnitStorageAttachments", args, &results)
	if err != nil {
		return nil, err
	}
	if len(results.Results) != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}
	w := apiwatcher.NewStringsWatcher(sa.facade.RawAPICaller(), result)
	return w, nil
}
339584
func (m *Machine) WatchContainers() (watcher.StringsWatcher, error) {
	var result params.StringsWatchResult
	arg := params.Entity{Tag: m.tag.String()}
	err := m.facade.FacadeCall("WatchContainers", arg, &result)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if result.Error != nil {
		return nil, result.Error
	}
	return apiwatcher.NewStringsWatcher(m.facade.RawAPICaller(), result), nil
}
537341
func (cacher *txSenderCacher) cache() {
	for task := range cacher.tasks {
		for i := 0; i < len(task.txs); i += task.inc {
			types.Sender(task.signer, task.txs[i])
		}
	}
}
1215707
func newHeuristicWatchCacheSizes(expectedRAMCapacityMB int) map[schema.GroupResource]int {
	// TODO: Revisit this heuristic, copied from upstream
	clusterSize := expectedRAMCapacityMB / 60

	// default enable watch caches for resources that will have a high number of clients accessing it
	// and where the write rate may be significant
	watchCacheSizes := make(map[schema.GroupResource]int)
	watchCacheSizes[schema.GroupResource{Group: "network.openshift.io", Resource: "hostsubnets"}] = maxInt(5*clusterSize, 100)
	watchCacheSizes[schema.GroupResource{Group: "network.openshift.io", Resource: "netnamespaces"}] = maxInt(5*clusterSize, 100)
	watchCacheSizes[schema.GroupResource{Group: "network.openshift.io", Resource: "egressnetworkpolicies"}] = maxInt(10*clusterSize, 100)
	return watchCacheSizes
}
1115266
func (st *State) watchEnqueuedActionsFilteredBy(receivers ...ActionReceiver) StringsWatcher {
	return newCollectionWatcher(st, colWCfg{
		col:    actionNotificationsC,
		filter: makeIdFilter(st, actionMarker, receivers...),
		idconv: actionNotificationIdToActionId,
	})
}
1309836
func (h *Holder) monitorCacheFlush() {
	ticker := time.NewTicker(h.cacheFlushInterval)
	defer ticker.Stop()

	for {
		select {
		case <-h.closing:
			return
		case <-ticker.C:
			h.flushCaches()
		}
	}
}
897131
func (u *Unit) Watch() NotifyWatcher {
	return newEntityWatcher(u.st, unitsC, u.doc.DocID)
}
269057
func (f FakeWatcher) Watch(api.ListOptions) (watch.Interface, error) {
	return f.IFace, f.Err
}