if isEventedPLEGInUse() {
		return
	}
	setEventedPLEGUsage(true)
	e.stopCh = make(chan struct{})
	e.stopCacheUpdateCh = make(chan struct{})
	go wait.Until(e.watchEventsChannel, 0, e.stopCh)
	go wait.Until(e.updateGlobalCache, globalCacheUpdatePeriod, e.stopCacheUpdateCh)
}

// Stop stops the Evented PLEG
func (e *EventedPLEG) Stop() {
	e.runningMu.Lock()
	defer e.runningMu.Unlock()
	if !isEventedPLEGInUse() {

See the License for the specific language governing permissions and
limitations under the License.
*/

package admission

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

// TODO add a `WantsToRun` which takes a stopCh.  Might make it generic.

// WantsCloudConfig defines a function which sets CloudConfig for admission plugins that need it.
type WantsCloudConfig interface {
	SetCloudConfig([]byte)
}

// always return ErrWaitTimeout.
func Test_waitForWithEarlyClosing_waitFunc(t *testing.T) {
	stopCh := make(chan struct{})
	defer close(stopCh)

	ctx := ContextForChannel(stopCh)
	start := time.Now()
	err := waitForWithContext(ctx, func(ctx context.Context) <-chan struct{} {
		c := make(chan struct{})
		close(c)
		return c

	// serialize the operations on ServiceCIDRs
	go wait.Until(r.cidrWorker, r.workerLoopPeriod, stopCh)

	for i := 0; i < workers; i++ {
		go wait.Until(r.ipWorker, r.workerLoopPeriod, stopCh)
		go wait.Until(r.svcWorker, r.workerLoopPeriod, stopCh)
	}

	<-stopCh
}

// runOnce verifies the state of the ClusterIP allocations and returns an error if an unrecoverable problem occurs.

			handler, informerFactory, err := newHandlerForTest(mockClient)
			if err != nil {
				t.Fatalf("unexpected error initializing handler: %v", err)
			}
			stopCh := make(chan struct{})
			defer close(stopCh)
			informerFactory.Start(stopCh)

			handler.pluginConfig = &pluginapi.Configuration{Default: test.defaultClusterTolerations, Whitelist: test.clusterWhitelist}
			pod := test.pod

	cacher.reflector = reflector

	go cacher.dispatchEvents()
	go progressRequester.Run(stopCh)

	cacher.stopWg.Add(1)
	go func() {
		defer cacher.stopWg.Done()
		defer cacher.terminateAllWatchers()
		wait.Until(
			func() {
				if !cacher.isStopped() {
					cacher.startCaching(stopCh)
				}
			}, time.Second, stopCh,
		)
	}()

	return cacher, nil
}

	stopCh := make(chan struct{})

	env.runFunc = func() {
		go c.Run(&spec.Swagger{
			SwaggerProps: spec.SwaggerProps{
				Paths: &spec.Paths{
					Paths: map[string]spec.PathItem{
						"/apis/apiextensions.k8s.io/v1": {},
					},
				},
			},
		}, h, stopCh)
	}

	env.cleanFunc = func() {
		cancel()
		close(stopCh)
	}
	return env, ctx
}

	}
	return cast
}

func (n *informerClient[T]) applyFilter(t T) bool {
	if n.filter == nil {
		return true
	}
	return n.filter(t)
}

func (n *informerClient[T]) Start(stopCh <-chan struct{}) {
	n.startInformer(stopCh)
}

func (n *writeClient[T]) Create(object T) (T, error) {
	api := kubeclient.GetWriteClient[T](n.client, object.GetNamespace())
	return api.Create(context.Background(), object, metav1.CreateOptions{})

		ExecSeatsGaugeVec:      metrics.PriorityLevelExecutionSeatsGaugeVec,
		QueueSetFactory:        cts,
	})

	stopCh, controllerCompletedCh := make(chan struct{}), make(chan struct{})
	var controllerErr error

	informerFactory.Start(stopCh)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	status := informerFactory.WaitForCacheSync(ctx.Done())

func (c *Repair) RunUntil(onFirstSuccess func(), stopCh chan struct{}) {
	c.broadcaster.StartRecordingToSink(stopCh)
	defer c.broadcaster.Shutdown()

	var once sync.Once
	wait.Until(func() {
		if err := c.runOnce(); err != nil {
			runtime.HandleError(err)
			return
		}
		once.Do(onFirstSuccess)
	}, c.interval, stopCh)
}