// getWatchersByTopic returns a slice containing the filters that
// match a specific topic
558816
func (fs *Filters) getWatchersByTopic(topic TopicType) []*Filter {
	res := make([]*Filter, 0, len(fs.allTopicsMatcher))
	for watcher := range fs.allTopicsMatcher {
		res = append(res, watcher)
	}
	for watcher := range fs.topicMatcher[topic] {
		res = append(res, watcher)
	}
	return res
}
587525
func (c *Client) WatchApplications() (watcher.StringsWatcher, error) {
	var result params.StringsWatchResult
	if err := c.facade.FacadeCall("WatchApplications", nil, &result); err != nil {
		return nil, err
	}
	if err := result.Error; err != nil {
		return nil, result.Error
	}
	w := apiwatcher.NewStringsWatcher(c.facade.RawAPICaller(), result)
	return w, nil
}
1059790
func (g *GRPCServer) WatchEvents(watch *proto.Watch, stream proto.AuthService_WatchEventsServer) error {
	auth, err := g.authenticate(stream.Context())
	if err != nil {
		return trail.ToGRPC(err)
	}
	servicesWatch := services.Watch{
		Name: auth.User.GetName(),
	}
	for _, kind := range watch.Kinds {
		servicesWatch.Kinds = append(servicesWatch.Kinds, services.WatchKind{
			Name:        kind.Name,
			Kind:        kind.Kind,
			LoadSecrets: kind.LoadSecrets,
		})
	}
	watcher, err := auth.NewWatcher(stream.Context(), servicesWatch)
	if err != nil {
		return trail.ToGRPC(err)
	}
	defer watcher.Close()

	for {
		select {
		case <-stream.Context().Done():
			return nil
		case <-watcher.Done():
			return trail.ToGRPC(watcher.Error())
		case event := <-watcher.Events():
			out, err := eventToGRPC(event)
			if err != nil {
				return trail.ToGRPC(err)
			}
			if err := stream.Send(out); err != nil {
				return trail.ToGRPC(err)
			}
		}
	}
}
55555
func (m *Model) WatchActionResultsFilteredBy(receivers ...ActionReceiver) StringsWatcher {
	return newActionStatusWatcher(m.st, receivers, []ActionStatus{ActionCompleted, ActionCancelled, ActionFailed}...)
}
739009
func (_m *DBClient) GetAllProvisionWatchers() ([]models.ProvisionWatcher, error) {
	ret := _m.Called()

	var r0 []models.ProvisionWatcher
	if rf, ok := ret.Get(0).(func() []models.ProvisionWatcher); ok {
		r0 = rf()
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).([]models.ProvisionWatcher)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func() error); ok {
		r1 = rf()
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}
730260
func (n *node) PropagateWatchError(err error) {
	for _, ch := range n.watches {
		ch <- &topo.WatchData{
			Err: err,
		}
	}

	for _, c := range n.children {
		c.PropagateWatchError(err)
	}
}
394485
func (c *limitRanges) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Namespace(c.ns).
		Resource("limitRanges").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
548010
func (f *Facade) WatchActionNotifications(args params.Entities) params.StringsWatchResults {
	tagToActionReceiver := f.backend.TagToActionReceiverFn(f.backend.FindEntity)
	watchOne := common.WatchOneActionReceiverNotifications(tagToActionReceiver, f.resources.Register)
	return common.WatchActionNotifications(args, f.accessMachine, watchOne)
}
1046483
func (c *Client) WatchAll() (*AllWatcher, error) {
	var info params.AllWatcherId
	if err := c.facade.FacadeCall("WatchAll", nil, &info); err != nil {
		return nil, err
	}
	return NewAllWatcher(c.st, &info.AllWatcherId), nil
}
410230
func Watch(facade base.FacadeCaller, method string, tag names.Tag) (watcher.NotifyWatcher, error) {
	var results params.NotifyWatchResults
	args := params.Entities{
		Entities: []params.Entity{{Tag: tag.String()}},
	}
	err := facade.FacadeCall(method, args, &results)
	if err != nil {
		return nil, err
	}
	if len(results.Results) != 1 {
		return nil, fmt.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}
	return apiwatcher.NewNotifyWatcher(facade.RawAPICaller(), result), nil
}
511388
func (c *namespaces) Watch(opts api.ListOptions) (watch.Interface, error) {
	return c.r.Get().
		Prefix("watch").
		Resource("namespaces").
		VersionedParams(&opts, api.Scheme).
		Watch()
}
1304429
func (conn *ProtoConnection) Watch(msgClb func(messaging.ProtoMessage), topics ...string) error {
	return conn.ConsumeTopic(msgClb, topics...)
}
250828
func (u *UniterAPI) WatchActionNotifications(args params.Entities) (params.StringsWatchResults, error) {
	tagToActionReceiver := common.TagToActionReceiverFn(u.st.FindEntity)
	watchOne := common.WatchOneActionReceiverNotifications(tagToActionReceiver, u.resources.Register)
	canAccess, err := u.accessUnit()
	if err != nil {
		return params.StringsWatchResults{}, err
	}
	return common.WatchActionNotifications(args, canAccess, watchOne), nil
}
1308626
func (r *Repository) GetWatchersCount() int {
	if r == nil || r.WatchersCount == nil {
		return 0
	}
	return *r.WatchersCount
}
1054840
func (fw Watchers) WatchUnitManagedFilesystems(app names.ApplicationTag) state.StringsWatcher {
	w := &hostFilesystemsWatcher{
		stringsWatcherBase:     stringsWatcherBase{out: make(chan []string)},
		backend:                fw.Backend,
		changes:                set.NewStrings(),
		hostFilesystems:        fw.Backend.WatchUnitFilesystems(app),
		modelFilesystems:       fw.Backend.WatchModelFilesystems(),
		modelVolumeAttachments: fw.Backend.WatchModelVolumeAttachments(),
		modelVolumesAttached:   names.NewSet(),
		modelVolumeFilesystems: make(map[names.VolumeTag]names.FilesystemTag),
		hostMatch: func(tag names.Tag) (bool, error) {
			entity, err := names.UnitApplication(tag.Id())
			if err != nil {
				return false, errors.Trace(err)
			}
			return app.Id() == entity, nil
		},
	}
	w.tomb.Go(func() error {
		defer watcher.Stop(w.hostFilesystems, &w.tomb)
		defer watcher.Stop(w.modelFilesystems, &w.tomb)
		defer watcher.Stop(w.modelVolumeAttachments, &w.tomb)
		return w.loop()
	})
	return w
}
605808
func (m *NodeManager) GetAndWatchFullTree(notifications chan []byte) ([]byte, int, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	i := m.nextWatcherIndex
	m.nextWatcherIndex++

	result, err := m.toJSON(i)
	if err != nil {
		return nil, 0, err
	}

	// It worked, register the watcher.
	m.watchers[i] = notifications
	return result, i, nil
}
575446
func (t *Topic) WatchPartitions() (PartitionList, <-chan zk.Event, error) {
	value, _, c, err := t.kz.conn.GetW(t.metadataPath())
	if err != nil {
		return nil, nil, err
	}

	list, err := t.parsePartitions(value)
	return list, c, err
}