}
	}
	if !c.Client.RunAndWait(c.stop) {
		log.Warnf("remote cluster %s failed to sync", c.ID)
		return
	}
	for _, h := range syncers {
		if !kube.WaitForCacheSync("cluster"+string(c.ID), c.stop, h.HasSynced) {
			log.Warnf("remote cluster %s failed to sync handler", c.ID)
			return
		}
	}

	c.initialSync.Store(true)
}

// Stop closes the stop channel, if is safe to be called multi times.

	go w.startCaBundleWatcher(stopChan)
	w.webhooks.Start(stopChan)
	kubelib.WaitForCacheSync("webhook patcher", stopChan, w.webhooks.HasSynced)
	w.queue.Run(stopChan)
}

func (w *WebhookCertPatcher) HasSynced() bool {
	return w.queue.HasSynced()
}

// webhookPatchTask takes the result of patchMutatingWebhookConfig and modifies the result for use in task queue

			ttlc.updateNode(logger, old, newObj)
		},
		DeleteFunc: ttlc.deleteNode,
	})

	ttlc.nodeStore = listers.NewNodeLister(nodeInformer.Informer().GetIndexer())
	ttlc.hasSynced = nodeInformer.Informer().HasSynced

	return ttlc
}

type ttlBoundary struct {
	sizeMin    int
	sizeMax    int
	ttlSeconds int
}

var (
	ttlBoundaries = []ttlBoundary{

		i.inf.AddEventHandler(informerEventHandler[I](func(o Event[I], initialSync bool) {
			f([]Event[I]{o}, initialSync)
		}))
	}
	return pollSyncer{
		name: fmt.Sprintf("%v handler", i.name()),
		f:    i.inf.HasSynced,
	}
}

func informerEventHandler[I controllers.ComparableObject](handler func(o Event[I], initialSync bool)) cache.ResourceEventHandler {
	return controllers.EventHandler[I]{

	AppendWorkloadHandler(f func(*WorkloadInstance, Event))

	// Run until a signal is received
	Run(stop <-chan struct{})

	// HasSynced returns true after initial cache synchronization is complete
	HasSynced() bool
}

// AggregateController is a wrapper of Controller, it supports registering handlers of a specific cluster。
type AggregateController interface {
	Controller

	ObjectTransform func(obj any) (any, error)
}

// CrdWatcher exposes an interface to watch CRDs
type CrdWatcher interface {
	// HasSynced returns true once all existing state has been synced.
	HasSynced() bool
	// KnownOrCallback returns `true` immediately if the resource is known.
	// If it is not known, `false` is returned. If the resource is later added, the callback will be triggered.

	out := 0
	for _, clusterMap := range c.remoteClusters {
		out += len(clusterMap)
	}
	return out
}

func (c *ClusterStore) HasSynced() bool {
	c.RLock()
	defer c.RUnlock()
	for _, clusterMap := range c.remoteClusters {
		for _, cl := range clusterMap {
			if !cl.HasSynced() {
				log.Debugf("remote cluster %s registered informers have not been synced up yet", cl.ID)
				return false
			}
		}
	}

	"k8s.io/apiserver/pkg/admission"
)

type CELPolicyEvaluator interface {
	admission.InitializationValidator

	Validate(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) error
	HasSynced() bool
	Run(stopCh <-chan struct{})

			return
		}
		w.HandleMeshConfig(meshConfig)
	})

	go c.Run(stop)

	// Ensure the ConfigMap is initially loaded if present.
	if !client.WaitForCacheSync("configmap watcher", stop, c.HasSynced) {
		log.Error("failed to wait for cache sync")
	}
	return w
}

func AddUserMeshConfig(client kube.Client, watcher mesh.Watcher, namespace, key, userMeshConfig string, stop <-chan struct{}) {

		meshWatcher,
		stop,
	)
	kube.SetObjectFilter(client, discoveryNamespacesFilter)
	nc := NewNamespaceController(client, watcher)
	client.RunAndWait(stop)
	go nc.Run(stop)
	retry.UntilOrFail(t, nc.queue.HasSynced)

	expectedData := map[string]string{
		constants.CACertNamespaceConfigMapDataName: string(caBundle),
	}
	createNamespace(t, client.Kube(), "foo", nil)