public boolean dispatch(Connection<Message> connection, Command stopCommand) {
        Throwable failure = null;
        try {
            connection.dispatch(stopCommand);
            Result result = (Result) connection.receive();
            if (result instanceof Failure) {
                failure = ((Failure) result).getValue();
            }
            connection.dispatch(new Finished());

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.

    }

    def "runs action asynchronously"() {
        when:
        async {
            connection.run(action, handler)
            instant.dispatched
        }

        then:
        1 * actionExecuter.run(action) >> {
            thread.blockUntil.dispatched
            instant.actionStarted
            return "result"
        }
        1 * handler.onComplete("result") >> {
            instant.resultReceived

 * to the current closures chain and can potentially be intercepted. The implementation must ensure that all the closures in the scope are processed in a way that
 * ensures call interception if a call is dispatched to them.
 */
@NonNullApi
public interface InstrumentedClosuresTracker {

  @BeforeEach
  fun setUp(server: MockWebServer) {
    platform.assumeLoom()

    this.server = server

    client =
      clientTestRule.newClientBuilder()
        .dispatcher(Dispatcher(newVirtualThreadPerTaskExecutor()))
        .build()
  }

  private fun newVirtualThreadPerTaskExecutor(): ExecutorService {

    queue.enqueue(incrementingEvent(counters, listener, 3));
    queue.enqueue(incrementingEvent(counters, listener, 4));
    assertEquals(0, counters.size());
    queue.dispatch();
    assertEquals(multiset(listener, 4), counters);
  }

  public void testEnqueueAndDispatch_multipleListeners() {
    Object listener1 = new Object();
    ListenerCallQueue<Object> queue = new ListenerCallQueue<>();

    queue.enqueue(incrementingEvent(counters, listener, 3));
    queue.enqueue(incrementingEvent(counters, listener, 4));
    assertEquals(0, counters.size());
    queue.dispatch();
    assertEquals(multiset(listener, 4), counters);
  }

  public void testEnqueueAndDispatch_multipleListeners() {
    Object listener1 = new Object();
    ListenerCallQueue<Object> queue = new ListenerCallQueue<>();

            start {
                thread.blockUntil.dispatched
                instant.resultAvailable
                adaptedHandler.onComplete(null)
            }
        }
        handler.onComplete(null) >> {
            instant.resultReceived
        }

        when:
        async {
            executer.run(handler)
            instant.dispatched
            thread.blockUntil.resultReceived
        }

      this.taskRunner = okHttpClient.taskRunner
    }

    /**
     * Sets the dispatcher used to set policy and execute asynchronous requests. Must not be null.
     */
    fun dispatcher(dispatcher: Dispatcher) =
      apply {
        this.dispatcher = dispatcher
      }

    /**
     * Sets the connection pool used to recycle HTTP and HTTPS connections.
     *

import kotlin.contracts.InvocationKind
import kotlin.contracts.contract
import okhttp3.Dispatcher
import okhttp3.internal.http2.Http2Connection
import okhttp3.internal.http2.Http2Stream
import okhttp3.internal.http2.Http2Writer

/**
 * Centralisation of central locks according to docs/contribute/concurrency.md
 */
internal object Locks {
  inline fun <T> Dispatcher.withLock(action: () -> T): T {