Name: fmt.Sprintf("%d", eds),
						}),
					})
				}
			}
		}()

		done := make(chan struct{})
		mu := sync.RWMutex{}
		go func() {
			for {
				con, info, shuttingdown := p.Dequeue()
				if shuttingdown {
					return
				}
				for eds := range model.ConfigNamesOfKind(info.ConfigsUpdated, kind.ServiceEntry) {
					mu.Lock()
					delete(expected, key(con, eds))
					mu.Unlock()
				}

	startedInformers map[reflect.Type]bool
	// wg tracks how many goroutines were started.
	wg sync.WaitGroup
	// shuttingDown is true when Shutdown has been called. It may still be running
	// because it needs to wait for goroutines.
	shuttingDown bool
}

// WithCustomResyncConfig sets a custom resync period for the specified informer types.

	startedInformers map[reflect.Type]bool
	// wg tracks how many goroutines were started.
	wg sync.WaitGroup
	// shuttingDown is true when Shutdown has been called. It may still be running
	// because it needs to wait for goroutines.
	shuttingDown bool
}

// WithCustomResyncConfig sets a custom resync period for the specified informer types.

	q.metrics.latency.Record(time.Since(task.enqueueTime).Seconds())

	return task, false
}

func (q *queueImpl) processNextItem() bool {
	// Wait until there is a new item in the queue
	task, shuttingdown := q.get()
	if shuttingdown {
		return false
	}

	// Run the task.
	if err := task.task(); err != nil {
		delay := q.delay
		log.Infof("Work item handle failed (%v), retry after delay %v", err, delay)

			// This limits the number of pushes that can happen concurrently
			semaphore <- struct{}{}

			// Get the next proxy to push. This will block if there are no updates required.
			client, push, shuttingdown := queue.Dequeue()
			if shuttingdown {
				return
			}
			recordPushTriggers(push.Reason)
			// Signals that a push is done by reading from the semaphore, allowing another send on it.
			doneFunc := func() {

	}

	var serverShuttingDownCh <-chan struct{}
	if signals := apirequest.ServerShutdownSignalFrom(req.Context()); signals != nil {
		serverShuttingDownCh = signals.ShuttingDown()
	}

	server := &WatchServer{
		Watching: watcher,
		Scope:    scope,

		UseTextFraming:  useTextFraming,
		MediaType:       mediaType,
		Framer:          framer,
		Encoder:         encoder,

	if s.ReadHeaderTimeout != 0 {
		return s.ReadHeaderTimeout
	}
	return s.ReadTimeout
}

func (s *Server) doKeepAlives() bool {
	return !s.disableKeepAlives.Load() && !s.shuttingDown()
}

func (s *Server) shuttingDown() bool {
	return s.inShutdown.Load()
}

// SetKeepAlivesEnabled controls whether HTTP keep-alives are enabled.
// By default, keep-alives are always enabled. Only very

		// this simulates a watch well enough for our test
		signals := apirequest.ServerShutdownSignalFrom(req.Context())
		if signals == nil {
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		<-signals.ShuttingDown()
		w.WriteHeader(http.StatusOK)
	})
	s.Handler.NonGoRestfulMux.Handle("/apis/watches.group/v1/namespaces/foo/bar", handler)
	return inflight
}

func (t *trackingWorkqueue[T]) ShutDown() {
	t.limiter.ShutDown()
}
func (t *trackingWorkqueue[T]) ShutDownWithDrain() {
	t.limiter.ShutDownWithDrain()
}
func (t *trackingWorkqueue[T]) ShuttingDown() bool {
	return t.limiter.ShuttingDown()
}

func (t *trackingWorkqueue[T]) queue(item T) {
	if _, queued := t.pendingMap[item]; queued {
		// fmt.Printf("already queued: %#v\n", item)
		return
	}

SweeperPool { private static final boolean DEBUG = 0; private transient SweeperPool$Sweeper sweeper; private transient int maxSize; private transient int minSize; private int triggerSize; private java.util.ArrayList pooledObjects; private boolean shuttingDown; public void SweeperPool(int, int, int, int, int); private int saneConvert(int); public synchronized Object get(); public synchronized boolean put(Object); public synchronized int getSize(); public synchronized void dispose(); public synchronized...