taskFaker.close()
    routePlanner.close()
  }

  @Test
  fun takeConnectedConnection() {
    val plan0 = routePlanner.addPlan()
    plan0.connectState = TLS_CONNECTED

    taskRunner.newQueue().execute("connect") {
      val result0 = finder.find()
      assertThat(result0).isEqualTo(plan0.connection)
    }

    taskFaker.runTasks()
    assertEvents(
      "take plan 0",
    )

) {
  internal val keepAliveDurationNs: Long = timeUnit.toNanos(keepAliveDuration)

  @Volatile
  private var addressStates: Map<Address, AddressState> = mapOf()

  private val cleanupQueue: TaskQueue = taskRunner.newQueue()
  private val cleanupTask =
    object : Task("$okHttpName ConnectionPool connection closer") {
      override fun runOnce(): Long = closeConnections(System.nanoTime())
    }

  /** Asynchronously writes frames to the outgoing socket. */
  private val writerQueue = taskRunner.newQueue()

  /** Ensures push promise callbacks events are sent in order per stream. */
  private val pushQueue = taskRunner.newQueue()

  /** Notifies the listener of settings changes. */
  private val settingsListenerQueue = taskRunner.newQueue()

  /** User code to run in response to push promise events. */

  }

  private void growIfNeeded() {
    if (size > queue.length) {
      int newCapacity = calculateNewCapacity();
      Object[] newQueue = new Object[newCapacity];
      arraycopy(queue, 0, newQueue, 0, queue.length);
      queue = newQueue;
    }
  }

  /** Returns ~2x the old capacity if small; ~1.5x otherwise. */
  private int calculateNewCapacity() {
    int oldCapacity = queue.length;

  private val taskRunner = taskFaker.taskRunner
  private val log = mutableListOf<String>()
  private val redQueue = taskRunner.newQueue()
  private val blueQueue = taskRunner.newQueue()
  private val greenQueue = taskRunner.newQueue()

  @AfterEach
  internal fun tearDown() {
    taskFaker.close()
  }

  @Test fun executeDelayed() {
    redQueue.execute("task", 100.µs) {

  }

  private void growIfNeeded() {
    if (size > queue.length) {
      int newCapacity = calculateNewCapacity();
      Object[] newQueue = new Object[newCapacity];
      arraycopy(queue, 0, newQueue, 0, queue.length);
      queue = newQueue;
    }
  }

  /** Returns ~2x the old capacity if small; ~1.5x otherwise. */
  private int calculateNewCapacity() {
    int oldCapacity = queue.length;

    lock.assertHeld()
    if (executeCallCount > runCallCount) return // A thread is still starting.

    executeCallCount++
    backend.execute(this@TaskRunner, runnable)
  }

  fun newQueue(): TaskQueue {
    val name = lock.withLock { nextQueueName++ }
    return TaskQueue(this, "Q$name")
  }

  /**
   * Returns a snapshot of queues that currently have tasks scheduled. The task runner does not

	c.pods = kclient.NewFiltered[*corev1.Pod](client, kclient.Filter{
		LabelSelector: cfg.LabelSelectors,
		FieldSelector: strings.Join(fieldSelectors, ","),
	})
	c.queue = controllers.NewQueue("repair pods",
		controllers.WithReconciler(c.Reconcile),
		controllers.WithMaxAttempts(5))
	c.pods.AddEventHandler(controllers.ObjectHandler(c.queue.AddObject))

	return c, nil
}

	s := &InformerHandlers{ctx: ctx, dataplane: dataplane, systemNamespace: systemNamespace}
	s.queue = controllers.NewQueue("ambient",
		controllers.WithGenericReconciler(s.reconcile),
		controllers.WithMaxAttempts(5),
	)
	// We only need to handle pods on our node

    // Reuse if the user specified a port
    serverSocket!!.reuseAddress = inetSocketAddress.port != 0
    serverSocket!!.bind(inetSocketAddress, 50)

    portField = serverSocket!!.localPort

    taskRunner.newQueue().execute("MockWebServer $portField", cancelable = false) {
      try {
        logger.fine("$this starting to accept connections")
        acceptConnections()
      } catch (e: Throwable) {