* #enqueue enqueueing} and then {@linkplain #dispatch dispatching} events.
 *
 * <p>The API of this class is designed to make it easy to achieve the following properties
 *
 * <ul>
 *   <li>Multiple events for the same listener are never dispatched concurrently.
 *   <li>Events for the different listeners are dispatched concurrently.
 *   <li>All events for a given listener dispatch on the provided executor.

- Watch requests that are delegated to aggregated API servers no longer reserve concurrency units (seats) in the API Priority and Fairness dispatcher for their entire duration. ([#105511](https://github.com/kubernetes/kubernetes/pull/105511), [@benluddy](https://github.com/benluddy))

  - apiserver_flowcontrol_current_r: R(the time of the last change in state of the queues)
  - apiserver_flowcontrol_dispatch_r: R(the time of the latest request dispatch)
  - apiserver_flowcontrol_latest_s: S(the request last dispatched) = R(when that request starts executing in the virtual world)
  - apiserver_flowcontrol_next_s_bounds: min and max next S among non-empty queues

import java.io.IOException
import java.util.concurrent.TimeUnit
import kotlin.test.assertFailsWith
import kotlin.time.Duration.Companion.seconds
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.job
import kotlinx.coroutines.supervisorScope
import kotlinx.coroutines.test.runTest

So we have a dedicated thread for every socket that just reads frames and dispatches them.

The reader thread must never run application-layer code. Otherwise one slow stream can hold up the entire connection.

import org.apache.maven.eventspy.EventSpy;
import org.apache.maven.execution.ExecutionListener;
import org.eclipse.aether.RepositoryListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dispatches callbacks to all registered EventSpies.
 * @since 3.0.2
 */
@Named
@Singleton
public class EventSpyDispatcher {
    private final Logger logger = LoggerFactory.getLogger(getClass());

        assertTrue(heartbeatCalled.get());
        assertTrue(cleanupCalled.get());
    }

    @Test
    public void test_poll_dispatchesEvents() {
        // Verify that poll fetches events and dispatches them
        final AtomicInteger dispatchCount = new AtomicInteger(0);
        final CoordinatorHelper helper = new CoordinatorHelper() {
            @Override
            public void sendHeartbeat() {

    event: Annotated[
        str | None,
        Doc(
            """
            Optional event type name.

            Maps to `addEventListener(event, ...)` on the browser. When omitted,
            the browser dispatches on the generic `message` event.
            """
        ),
    ] = None
    id: Annotated[
        str | None,
        AfterValidator(_check_id_no_null),
        Doc(
            """

    this.bus = bus;
    this.target = checkNotNull(target);
    this.method = method;
    method.setAccessible(true);

    this.executor = bus.executor();
  }

  /** Dispatches {@code event} to this subscriber using the proper executor. */
  final void dispatchEvent(Object event) {
    executor.execute(
        () -> {
          try {
            invokeSubscriberMethod(event);

    // Now we find the task in our Executor and explicitly activate it.
    List<Runnable> tasks = executor.getTasks();
    assertEquals("One event dispatch task should be queued.", 1, tasks.size());

    tasks.get(0).run();

    assertEquals("One event should be delivered.", 1, events.size());