// removeWatchedUnit stops watching the unit, and all subordinates for this unit
436297
func (w *machineUnitsWatcher) removeWatchedUnit(unitName string, doc unitDoc, pending []string) ([]string, error) {
	logger.Tracef("machineUnitsWatcher removing unit %q for life %q", doc.Name, doc.Life)
	life, known := w.known[unitName]
	// Unit was removed or unassigned from w.machine
	if known {
		delete(w.known, unitName)
		docID := w.backend.docID(unitName)
		w.watcher.Unwatch(unitsC, docID, w.in)
		if life != Dead && !hasString(pending, unitName) {
			pending = append(pending, unitName)
		}
		for _, subunitName := range doc.Subordinates {
			if sublife, subknown := w.known[subunitName]; subknown {
				delete(w.known, subunitName)
				w.watcher.Unwatch(unitsC, w.backend.docID(subunitName), w.in)
				if sublife != Dead && !hasString(pending, subunitName) {
					pending = append(pending, subunitName)
				}
			}
		}
	}
	return pending, nil
}
1234275
func (u *Unit) WatchStorage() (watcher.StringsWatcher, error) {
	return u.st.WatchUnitStorageAttachments(u.tag)
}
223646
func (g *Cloud) watchClusterID(stop <-chan struct{}) {
	g.ClusterID = ClusterID{
		cfgMapKey: fmt.Sprintf("%v/%v", UIDNamespace, UIDConfigMapName),
		client:    g.client,
	}

	mapEventHandler := cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			m, ok := obj.(*v1.ConfigMap)
			if !ok || m == nil {
				klog.Errorf("Expected v1.ConfigMap, item=%+v, typeIsOk=%v", obj, ok)
				return
			}
			if m.Namespace != UIDNamespace ||
				m.Name != UIDConfigMapName {
				return
			}

			klog.V(4).Infof("Observed new configmap for clusteriD: %v, %v; setting local values", m.Name, m.Data)
			g.ClusterID.update(m)
		},
		UpdateFunc: func(old, cur interface{}) {
			m, ok := cur.(*v1.ConfigMap)
			if !ok || m == nil {
				klog.Errorf("Expected v1.ConfigMap, item=%+v, typeIsOk=%v", cur, ok)
				return
			}

			if m.Namespace != UIDNamespace ||
				m.Name != UIDConfigMapName {
				return
			}

			if reflect.DeepEqual(old, cur) {
				return
			}

			klog.V(4).Infof("Observed updated configmap for clusteriD %v, %v; setting local values", m.Name, m.Data)
			g.ClusterID.update(m)
		},
	}

	listerWatcher := cache.NewListWatchFromClient(g.ClusterID.client.CoreV1().RESTClient(), "configmaps", UIDNamespace, fields.Everything())
	var controller cache.Controller
	g.ClusterID.store, controller = cache.NewInformer(newSingleObjectListerWatcher(listerWatcher, UIDConfigMapName), &v1.ConfigMap{}, updateFuncFrequency, mapEventHandler)

	controller.Run(stop)
}
206707
func (u *UniterAPIV8) WatchUnitAddresses(args params.Entities) (params.NotifyWatchResults, error) {
	result := params.NotifyWatchResults{
		Results: make([]params.NotifyWatchResult, len(args.Entities)),
	}
	canAccess, err := u.accessUnit()
	if err != nil {
		return params.NotifyWatchResults{}, err
	}
	for i, entity := range args.Entities {
		unit, err := names.ParseUnitTag(entity.Tag)
		if err != nil {
			result.Results[i].Error = common.ServerError(common.ErrPerm)
			continue
		}
		err = common.ErrPerm
		watcherId := ""
		if canAccess(unit) {
			watcherId, err = u.watchOneUnitAddresses(unit)
		}
		result.Results[i].NotifyWatcherId = watcherId
		result.Results[i].Error = common.ServerError(err)
	}
	return result, nil
}
1320233
func (fw *Firewaller) forgetUnit(unitd *unitData) {
	applicationd := unitd.applicationd
	machined := unitd.machined

	// If it's the last unit in the application, we'll need to stop the applicationd.
	stoppedApplication := false
	if len(applicationd.unitds) == 1 {
		if _, found := applicationd.unitds[unitd.tag]; found {
			// Unusually, it's fine to ignore this error, because we know the
			// applicationd is being tracked in fw.catacomb. But we do still want
			// to wait until the watch loop has stopped before we nuke the last
			// data and return.
			_ = worker.Stop(applicationd)
			stoppedApplication = true
		}
	}

	// Clean up after stopping.
	delete(fw.unitds, unitd.tag)
	delete(machined.unitds, unitd.tag)
	delete(applicationd.unitds, unitd.tag)
	logger.Debugf("stopped watching %q", unitd.tag)
	if stoppedApplication {
		applicationTag := applicationd.application.Tag()
		delete(fw.applicationids, applicationTag)
		logger.Debugf("stopped watching %q", applicationTag)
	}
}
1169579
func (_PublicResolver *PublicResolverFilterer) WatchABIChanged(opts *bind.WatchOpts, sink chan<- *PublicResolverABIChanged, node [][32]byte, contentType []*big.Int) (event.Subscription, error) {

	var nodeRule []interface{}
	for _, nodeItem := range node {
		nodeRule = append(nodeRule, nodeItem)
	}
	var contentTypeRule []interface{}
	for _, contentTypeItem := range contentType {
		contentTypeRule = append(contentTypeRule, contentTypeItem)
	}

	logs, sub, err := _PublicResolver.contract.WatchLogs(opts, "ABIChanged", nodeRule, contentTypeRule)
	if err != nil {
		return nil, err
	}
	return event.NewSubscription(func(quit <-chan struct{}) error {
		defer sub.Unsubscribe()
		for {
			select {
			case log := <-logs:
				// New log arrived, parse the event and forward to the user
				event := new(PublicResolverABIChanged)
				if err := _PublicResolver.contract.UnpackLog(event, "ABIChanged", log); err != nil {
					return err
				}
				event.Raw = log

				select {
				case sink <- event:
				case err := <-sub.Err():
					return err
				case <-quit:
					return nil
				}
			case err := <-sub.Err():
				return err
			case <-quit:
				return nil
			}
		}
	}), nil
}
564225
func (r *Relation) WatchUnits(appName string) (RelationUnitsWatcher, error) {
	return r.watchUnits(appName, false)
}
692947
func (w *Watcher) RemovePin(p uint) {
	pin := Pin{
		Number: p,
	}
	w.cmdChan <- watcherCmd{
		pin:    pin,
		action: watcherRemove,
	}
}
238718
func (k *kubernetesClient) WatchService(appName string) (watcher.NotifyWatcher, error) {
	// Application may be a statefulset or deployment. It may not have
	// been set up when the watcher is started so we don't know which it
	// is ahead of time. So use a multi-watcher to cover both cases.
	statefulsets := k.client().AppsV1().StatefulSets(k.namespace)
	sswatcher, err := statefulsets.Watch(v1.ListOptions{
		LabelSelector: applicationSelector(appName),
		Watch:         true,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	w1, err := k.newWatcher(sswatcher, appName, k.clock)
	if err != nil {
		return nil, errors.Trace(err)
	}

	deployments := k.client().AppsV1().Deployments(k.namespace)
	dwatcher, err := deployments.Watch(v1.ListOptions{
		LabelSelector: applicationSelector(appName),
		Watch:         true,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	w2, err := k.newWatcher(dwatcher, appName, k.clock)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return watcher.NewMultiNotifyWatcher(w1, w2), nil
}
519825
func (w *Watcher) remove(d *structs.Deployment) {
	w.l.Lock()
	defer w.l.Unlock()

	// Not enabled so no-op
	if !w.enabled {
		return
	}

	if watcher, ok := w.watchers[d.ID]; ok {
		watcher.StopWatch()
		delete(w.watchers, d.ID)
	}
}
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
}
536662
func (w *Watcher) removePin(p Pin) {
	// we don't index by pin, so go looking
	for fd, pin := range w.pins {
		if pin.Number == p.Number {
			// found pin
			w.removeFd(fd)
			return
		}
	}
}
107918
func (u *UnitUpgraderAPI) WatchAPIVersion(args params.Entities) (params.NotifyWatchResults, error) {
	result := params.NotifyWatchResults{
		Results: make([]params.NotifyWatchResult, len(args.Entities)),
	}
	for i, agent := range args.Entities {
		tag, err := names.ParseTag(agent.Tag)
		if err != nil {
			result.Results[i].Error = common.ServerError(common.ErrPerm)
			continue
		}
		err = common.ErrPerm
		if u.authorizer.AuthOwner(tag) {
			var watcherId string
			watcherId, err = u.watchAssignedMachine(tag)
			if err == nil {
				result.Results[i].NotifyWatcherId = watcherId
			}
		}
		result.Results[i].Error = common.ServerError(err)
	}
	return result, nil
}
544937
func (c *client) removeReplySubTimeout(sub *subscription) {
	// Remove any reply sub timer if it exists.
	if c.route == nil || c.route.replySubs == nil {
		return
	}
	if t, ok := c.route.replySubs[sub]; ok {
		t.Stop()
		delete(c.route.replySubs, sub)
	}
}
319323
func (a *API) WatchUnitAssignments() (params.StringsWatchResult, error) {
	watch := a.st.WatchForUnitAssignment()
	if changes, ok := <-watch.Changes(); ok {
		return params.StringsWatchResult{
			StringsWatcherId: a.res.Register(watch),
			Changes:          changes,
		}, nil
	}
	return params.StringsWatchResult{}, watcher.EnsureErr(watch)
}
1211315
func (m *Monitor) removeListener(ml listener.MonitorListener) {
	m.Lock()
	defer m.Unlock()

	delete(m.listeners, ml)
	log.WithFields(logrus.Fields{
		"count.listener": len(m.listeners),
		"version":        ml.Version(),
	}).Debug("Removed listener")

	// If this was the final listener, shutdown the perf reader and unmap our
	// ring buffer readers. This tells the kernel to not emit this data.
	// Note: it is critical to hold the lock and check the number of listeners.
	// This guards against an older generation listener calling the
	// current generation perfReaderCancel
	if len(m.listeners) == 0 {
		m.perfReaderCancel()
	}
}
1219375
func (s *StorageAPI) WatchUnitStorageAttachments(args params.Entities) (params.StringsWatchResults, error) {
	canAccess, err := s.accessUnit()
	if err != nil {
		return params.StringsWatchResults{}, err
	}
	results := params.StringsWatchResults{
		Results: make([]params.StringsWatchResult, len(args.Entities)),
	}
	for i, entity := range args.Entities {
		result, err := s.watchOneUnitStorageAttachments(entity.Tag, canAccess)
		if err == nil {
			results.Results[i] = result
		}
		results.Results[i].Error = common.ServerError(err)
	}
	return results, nil
}