c.services.AddEventHandler(controllers.EventHandler[controllers.Object]{AddFunc: c.queue.AddObject})

	return c
}

func (c *autoServiceExportController) Run(stopCh <-chan struct{}) {
	kube.WaitForCacheSync("auto service export", stopCh, c.services.HasSynced)
	c.queue.Run(stopCh)
	c.services.ShutdownHandlers()
}

func (c *autoServiceExportController) logPrefix() string {
	return "AutoServiceExport (cluster=" + c.ClusterID.String() + ") "

	_ = c.RunOnce()

	// doesn't matter what workers say, only start one.
	go wait.Until(c.runWorker, time.Second, stopCh)

	// start timer that rechecks every minute, just in case.  this also serves to prime the controller quickly.
	go wait.Until(func() {
		c.Enqueue()
	}, 1*time.Minute, stopCh)

	<-stopCh
}

func (c *DynamicServingCertificateController) runWorker() {
	for c.processNextWorkItem() {
	}
}

	}
	return l
}

func newETCD3HealthCheck(c storagebackend.Config, stopCh <-chan struct{}) (func() error, error) {
	timeout := storagebackend.DefaultHealthcheckTimeout
	if c.HealthcheckTimeout != time.Duration(0) {
		timeout = c.HealthcheckTimeout
	}
	return newETCD3Check(c, timeout, stopCh)
}

func newETCD3ReadyCheck(c storagebackend.Config, stopCh <-chan struct{}) (func() error, error) {

	})
}

func (c *DiscoveryController) Run(stopCh <-chan struct{}, synchedCh chan<- struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()
	defer klog.Info("Shutting down DiscoveryController")

	klog.Info("Starting DiscoveryController")

	if !cache.WaitForCacheSync(stopCh, c.crdsSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))

// limitations under the License.

package concurrent

import (
	"time"

	"istio.io/istio/pkg/util/sets"
)

type Debouncer[T comparable] struct{}

func (d *Debouncer[T]) Run(ch chan T, stopCh <-chan struct{}, debounceMinInterval, debounceMaxInterval time.Duration, pushFn func(sets.Set[T])) {
	var timeChan <-chan time.Time
	var startDebounce time.Time
	var lastConfigUpdateTime time.Time

	pushCounter := 0

func (c *KubernetesAPIApprovalPolicyConformantConditionController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting KubernetesAPIApprovalPolicyConformantConditionController")
	defer klog.Infof("Shutting down KubernetesAPIApprovalPolicyConformantConditionController")

	if !cache.WaitForCacheSync(stopCh, c.crdSynced) {
		return
	}

	for i := 0; i < workers; i++ {

	// into channels.
	go func(stopCh <-chan struct{}) {
		for {
			nodeUpdate, shutdown := tc.nodeUpdateQueue.Get()
			if shutdown {
				break
			}
			hash := hash(nodeUpdate.nodeName, UpdateWorkerSize)
			select {
			case <-stopCh:
				tc.nodeUpdateQueue.Done(nodeUpdate)
				return
			case tc.nodeUpdateChannels[hash] <- nodeUpdate:

	// Register the metrics we defined above with prometheus.
	customMetricRegistry.MustRegister(etcdVersion)

	// Spawn threads for periodically scraping etcd version metrics.
	stopCh := make(chan struct{})
	defer close(stopCh)
	go getVersionPeriodically(stopCh)

	// Serve our metrics on listenAddress/metricsPath.
	klog.Infof("Listening on: %v", listenAddress)
	http.Handle(metricsPath, metrics.HandlerFor(gatherer, metrics.HandlerOpts{}))

	c, informers, err := f.newController(ctx)
	if err != nil {
		f.t.Fatalf("error creating Deployment controller: %v", err)
	}
	if startInformers {
		stopCh := make(chan struct{})
		defer close(stopCh)
		informers.Start(stopCh)
	}

	err = c.syncDeployment(ctx, deploymentName)
	if !expectError && err != nil {
		f.t.Errorf("error syncing deployment: %v", err)
	} else if expectError && err == nil {

		if features.LocalClusterSecretWatcher && features.ExternalIstiod {
			c.secrets.Start(stopCh)
		}
		if !kube.WaitForCacheSync("multicluster remote secrets", stopCh, c.secrets.HasSynced) {
			return
		}
		log.Infof("multicluster remote secrets controller cache synced in %v", time.Since(t0))
		c.queue.Run(stopCh)
		c.handleDelete(c.configClusterID)
	}()
	return nil
}