default:
		close(sw.done)
		sw.source.Close()
	}
}

// receive reads result from the decoder in a loop and sends down the result channel.
func (sw *StreamWatcher) receive() {
	defer utilruntime.HandleCrash()
	defer close(sw.result)
	defer sw.Stop()
	for {
		action, obj, err := sw.source.Decode()
		if err != nil {
			switch err {
			case io.EOF:
				// watch closed normally
			case io.ErrUnexpectedEOF:

	if err != nil {
		return nil, fmt.Errorf("failed to create connection to %s, error: %v", endpoint, err)
	}

	s.kmsClient = kmsapi.NewKeyManagementServiceClient(s.connection)

	go func() {
		defer utilruntime.HandleCrash()

		<-ctx.Done()
		_ = s.connection.Close()
	}()

	return s, nil
}

func (g *gRPCService) checkAPIVersion(ctx context.Context) error {
	g.mux.Lock()
	defer g.mux.Unlock()

	})

	return c
}

func (c *Controller) enqueue() {
	c.queue.Add(queueKey)
}

// Run starts the controller sync loop.
func (c *Controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Info("Starting legacy_token_tracking_controller")
	defer klog.Infof("Shutting down legacy_token_tracking_controller")

	go c.configMapInformer.Run(stopCh)

	return ic, nil
}

// Run starts an asynchronous loop that monitors the status of cluster nodes.
func (nc *Controller) Run(ctx context.Context) {
	defer utilruntime.HandleCrash()

	// Start event processing pipeline.
	nc.eventBroadcaster.StartStructuredLogging(3)
	nc.eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: nc.kubeClient.CoreV1().Events("")})

	return cc
}

// Run the main goroutine responsible for watching and syncing jobs.
func (cc *CertificateController) Run(ctx context.Context, workers int) {
	defer utilruntime.HandleCrash()
	defer cc.queue.ShutDown()

	logger := klog.FromContext(ctx)
	logger.Info("Starting certificate controller", "name", cc.name)
	defer logger.Info("Shutting down certificate controller", "name", cc.name)

	grm.operations[operationName] = operation{
		operationPending: true,
		expBackoff:       existingOp.expBackoff,
	}
	go func() (err error) {
		// Handle unhandled panics (very unlikely)
		defer k8sRuntime.HandleCrash()
		// Handle completion of and error, if any, from operationFunc()
		defer grm.operationComplete(operationName, &err)
		// Handle panic, if any, from operationFunc()
		defer k8sRuntime.RecoverFromPanic(&err)

	nsListerSynced cache.InformerSynced

	queue workqueue.TypedRateLimitingInterface[string]
}

// Run starts process
func (c *Publisher) Run(ctx context.Context, workers int) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	logger := klog.FromContext(ctx)
	logger.Info("Starting root CA cert publisher controller")
	defer logger.Info("Shutting down root CA cert publisher controller")

func ApplyClusterMerge(pctx networking.EnvoyFilter_PatchContext, efw *model.EnvoyFilterWrapper,
	c *cluster.Cluster, hosts []host.Name,
) (out *cluster.Cluster) {
	defer runtime.HandleCrash(runtime.LogPanic, func(any) {
		log.Errorf("clusters patch %s/%s caused panic, so the patches did not take effect", efw.Namespace, efw.Name)
		IncrementEnvoyFilterErrorMetric(Cluster)
	})

			},
		},
		options.SecretResync,
	)

	return e, nil
}

// Run runs controller loops and returns when they are done
func (tc *TokenCleaner) Run(ctx context.Context) {
	defer utilruntime.HandleCrash()
	defer tc.queue.ShutDown()

	logger := klog.FromContext(ctx)
	logger.Info("Starting token cleaner controller")
	defer logger.Info("Shutting down token cleaner controller")

		}
	}
	return false
}

// Run starts the controller and blocks until stopCh is closed.
func (c *ClusterRoleAggregationController) Run(ctx context.Context, workers int) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	logger := klog.FromContext(ctx)
	logger.Info("Starting ClusterRoleAggregator controller")
	defer logger.Info("Shutting down ClusterRoleAggregator controller")