// ResultChan implements Interface
func (w emptyWatch) ResultChan() <-chan Event {
	return chan Event(w)
}

// FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
type FakeWatcher struct {
	result  chan Event
	stopped bool
	sync.Mutex
}

func NewFake() *FakeWatcher {
	return &FakeWatcher{
		result: make(chan Event),
	}
}

func (h *StreamTypeHandler[Payload, Req, Resp]) RegisterNoPayload(m *Manager, handle func(ctx context.Context, in <-chan Req, out chan<- Resp) *RemoteErr, subroute ...string) error {
	h.WithPayload = false
	return h.register(m, func(ctx context.Context, p Payload, in <-chan Req, out chan<- Resp) *RemoteErr {
		return handle(ctx, in, out)
	}, subroute...)
}

	done := make(chan bool)
	AfterFunc(60*Second, func() { close(done) })

	// Set up a queuing goroutine to ping pong through the scheduler.
	inQ := make(chan func())
	outQ := make(chan func())

	defer close(inQ)

	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(outQ)
		var q []func()
		for {
			var sendTo chan func()
			var send func()
			if len(q) > 0 {

		if err := s.fileWatcher.Add(file); err != nil {
			return fmt.Errorf("could not watch %v: %v", file, err)
		}
	}
	s.addStartFunc("certificate rotation", func(stop <-chan struct{}) error {
		go func() {
			var keyCertTimerC <-chan time.Time
			for {
				select {
				case <-keyCertTimerC:
					keyCertTimerC = nil

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

// TagWatcher keeps track of the current tags and can notify watchers
// when the tags change.
type TagWatcher interface {
	Run(stopCh <-chan struct{})
	HasSynced() bool
	AddHandler(handler TagHandler)
	GetMyTags() sets.String
}

// TagHandler is a callback for when the tags revision change.
type TagHandler func(sets.String)

type tagWatcher struct {

	log.Infof("starting discovery service")
	// Implement EnvoyXdsServer grace shutdown
	s.addStartFunc("xds server", func(stop <-chan struct{}) error {
		log.Infof("Starting ADS server")
		s.XDSServer.Start(stop)
		return nil
	})
}

// Wait for the stop, and do cleanups
func (s *Server) waitForShutdown(stop <-chan struct{}) {
	go func() {
		<-stop
		close(s.internalStop)
		_ = s.fileWatcher.Close()

	"github.com/minio/minio/internal/logger"
	"github.com/minio/minio/internal/mcontext"
)

type apiConfig struct {
	mu sync.RWMutex

	requestsDeadline      time.Duration
	requestsPool          chan struct{}
	clusterDeadline       time.Duration
	listQuorum            string
	corsAllowOrigins      []string
	replicationPriority   string
	replicationMaxWorkers int
	transitionWorkers     int

				p.ShutdownDuration = 1 * time.Millisecond
			})
			g := NewWithT(t)
			s, err := NewServer(args, func(s *Server) {
				s.kubeClient = kube.NewFakeClient()
			})
			g.Expect(err).To(Succeed())
			stop := make(chan struct{})
			g.Expect(s.Start(stop)).To(Succeed())
			defer func() {
				close(stop)
				s.WaitUntilCompletion()
			}()

			if c.expNewCert {

	es := &expiryState{
		ctx:    ctx,
		objAPI: objAPI,
	}
	workers := make([]chan expiryOp, 0, n)
	es.workers.Store(&workers)
	es.ResizeWorkers(n)
	return es
}

func (es *expiryState) getWorkerCh(h uint64) chan<- expiryOp {
	w := es.workers.Load()
	if w == nil || len(*w) == 0 {
		return nil
	}
	workers := *w

	mrfMU      sync.Mutex
	resyncer   *replicationResyncer

	// workers:
	workers    []chan ReplicationWorkerOperation
	lrgworkers []chan ReplicationWorkerOperation

	// mrf:
	mrfWorkerKillCh chan struct{}
	mrfReplicaCh    chan ReplicationWorkerOperation
	mrfSaveCh       chan MRFReplicateEntry
	mrfStopCh       chan struct{}
	mrfWorkerSize   int
}