watcher, err := t.storage.(rest.Watcher).Watch(ctx, options)
			if err != nil {
				t.Errorf("unexpected error: %v, %v", err, action)
			}

			if err := emitFn(obj, action); err != nil {
				t.Errorf("unexpected error: %v", err)
			}

			select {
			case _, ok := <-watcher.ResultChan():
				if !ok {
					t.Errorf("watch channel should be open")
				}

// shouldn't we be doing this for both validating and mutating webhooks...?
func (c *Controller) startCaBundleWatcher(stop <-chan struct{}) {
	if c.o.CABundleWatcher == nil {
		return
	}
	id, watchCh := c.o.CABundleWatcher.AddWatcher()
	defer c.o.CABundleWatcher.RemoveWatcher(id)

	for {
		select {
		case <-watchCh:
			c.syncAll()
		case <-stop:

	c.queue.Add(workItemKey)

	w, err := fsnotify.NewWatcher()
	if err != nil {
		return fmt.Errorf("error creating fsnotify watcher: %v", err)
	}
	defer w.Close()

	if err := w.Add(c.certFile); err != nil {
		return fmt.Errorf("error adding watch for file %s: %v", c.certFile, err)
	}
	if err := w.Add(c.keyFile); err != nil {

	configMapNamespace string
	configMapName      string
	callback           func(*v1.ConfigMap)

	hasSynced atomic.Bool
}

// NewController returns a new ConfigMap watcher controller.
func NewController(client kube.Client, namespace, name string, callback func(*v1.ConfigMap)) *Controller {
	c := &Controller{
		configMapNamespace: namespace,
		configMapName:      name,
		callback:           callback,
	}

	// Setup Kubernetes watch filters
	// Because this relies on meshconfig, it needs to be outside initKubeClient
	if s.kubeClient != nil {
		// Build a namespace watcher. This must have no filter, since this is our input to the filter itself.
		namespaces := kclient.New[*corev1.Namespace](s.kubeClient)
		filter := namespace.NewDiscoveryNamespacesFilter(namespaces, s.environment.Watcher, s.internalStop)

	remoteCaCertPool   *x509.CertPool
	meshConfig         mesh.Watcher
}

var (
	trustBundleLog = log.RegisterScope("trustBundle", "Workload mTLS trust bundle logs")
	remoteTimeout  = 10 * time.Second
)

// NewTrustBundle returns a new trustbundle
func NewTrustBundle(remoteCaCertPool *x509.CertPool, meshConfig mesh.Watcher) *TrustBundle {
	var err error
	tb := &TrustBundle{

	c kube.Client,
	gvr schema.GroupVersionResource,
	informerType kubetypes.InformerType,
	filter Filter,
) Informer[T] {
	watcher := c.CrdWatcher()
	if watcher == nil {
		log.Fatalf("NewDelayedInformer called without a CrdWatcher enabled")
	}
	delay := newDelayedFilter(gvr, watcher)
	inf := func() informerfactory.StartableInformer {
		opts := ToOpts(c, gvr, filter)
		opts.InformerType = informerType

type Environment struct {
	// Discovery interface for listing services and instances.
	ServiceDiscovery

	// Config interface for listing routing rules
	ConfigStore

	// Watcher is the watcher for the mesh config (to be merged into the config store)
	mesh.Watcher

	// NetworksWatcher (loaded from a config map) provides information about the
	// set of networks inside a mesh and how to route to endpoints in each

				}
				return
			}
			if len(filter.Patches[networking.EnvoyFilter_CLUSTER]) != tt.expectedClusterPatches {
				t.Errorf("Expect %d envoy filter cluster patches, but got %d", tt.expectedClusterPatches, len(filter.Patches[networking.EnvoyFilter_CLUSTER]))
			}
			if len(filter.Patches[networking.EnvoyFilter_LISTENER]) != tt.expectedListenerPatches {

			if rev == "" {
				return nil
			}
			return []string{rev}
		}, controllers.ObjectHandler(p.queue.AddObject))
	return p
}

func (p *tagWatcher) Run(stopCh <-chan struct{}) {
	if !kube.WaitForCacheSync("tag watcher", stopCh, p.webhooks.HasSynced) {
		return
	}
	// Notify handlers of initial state
	p.notifyHandlers()
	p.queue.Run(stopCh)
}

// AddHandler registers a new handler for updates to tag changes.