return newTask.latch
    }
  }

  private class AwaitIdleTask : Task("$okHttpName awaitIdle", cancelable = false) {
    val latch = CountDownLatch(1)

    override fun runOnce(): Long {
      latch.countDown()
      return -1L
    }
  }

  /** Adds [task] to run in [delayNanos]. Returns true if the coordinator is impacted. */
  internal fun scheduleAndDecide(
    task: Task,
    delayNanos: Long,

    val latch = CountDownLatch(1)
    Thread {
      sleep(delayMillis)
      if (cancelMode == CANCEL) {
        call.cancel()
      } else {
        threadToCancel!!.interrupt()
      }
      latch.countDown()
    }.apply { start() }
    return latch
  }

  companion object {
    // The size of the socket buffers in bytes.
    private const val SOCKET_BUFFER_SIZE = 256 * 1024
  }
}

        .addNetworkInterceptor(
          Interceptor { chain ->
            try {
              chain.proceed(chain.request())
            } finally {
              firstConnectLatch.countDown()
            }
          },
        )
        .build()

    // Set up a same-connection retry.
    serverIpv4.enqueue(

              }
            });
    future.get(10, SECONDS);
    assertThrows(RejectedExecutionException.class, () -> executor.execute(Runnables.doNothing()));
    latch.countDown();
    ExecutionException expected =
        assertThrows(ExecutionException.class, () -> first.get(10, SECONDS));
    assertThat(expected).hasCauseThat().isInstanceOf(RejectedExecutionException.class);
  }

    ScheduledThreadPoolExecutor delegate =
        new ScheduledThreadPoolExecutor(1) {
          @Override
          protected void afterExecute(Runnable r, Throwable t) {
            completed.countDown();
          }
        };
    ListeningScheduledExecutorService service = listeningDecorator(delegate);
    ListenableFuture<Integer> future = service.schedule(Callables.returning(42), 1, MILLISECONDS);

    /*

        override fun onFailure(
          call: Call,
          e: IOException,
        ) {
          completionLatch.countDown()
        }

        override fun onResponse(
          call: Call,
          response: Response,
        ) {
          response.close()
          completionLatch.countDown()
        }
      }
    call.enqueue(callback)
    completionLatch.await()

    ScheduledThreadPoolExecutor delegate =
        new ScheduledThreadPoolExecutor(1) {
          @Override
          protected void afterExecute(Runnable r, Throwable t) {
            completed.countDown();
          }
        };
    ListeningScheduledExecutorService service = listeningDecorator(delegate);
    ListenableFuture<Integer> future = service.schedule(Callables.returning(42), 1, MILLISECONDS);

    /*

  @Param Impl impl;

  @Param({"1", "5", "10"})
  int numListeners;

  private final Runnable listener =
      new Runnable() {
        @Override
        public void run() {
          listenerLatch.countDown();
        }
      };

  @BeforeExperiment
  void setUp() throws Exception {
    executorService =
        new ThreadPoolExecutor(
            NUM_THREADS,
            NUM_THREADS,

    listener =
      object : RecordingEventListener() {
        override fun requestHeadersEnd(
          call: Call,
          request: Request,
        ) {
          requestFinished.countDown()
          super.responseHeadersStart(call)
        }
      }
    client =
      client.newBuilder()
        .dns(DoubleInetAddressDns())
        .eventListenerFactory(clientTestRule.wrap(listener))

          connection: Http2Connection,
          settings: Settings,
        ) {
          maxConcurrentStreams.set(settings.getMaxConcurrentStreams())
          maxConcurrentStreamsUpdated.countDown()
        }
      }
    val connection = connect(peer, IGNORE, listener)
    connection.withLock {
      assertThat(connection.peerSettings.getMaxConcurrentStreams()).isEqualTo(10)
    }