internal val connectionListener: ConnectionListener,
  private val exchangeFinderFactory: (RealConnectionPool, Address, ConnectionUser) -> ExchangeFinder,
) {
  internal val keepAliveDurationNs: Long = timeUnit.toNanos(keepAliveDuration)

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

  private val cleanupQueue: TaskQueue = taskRunner.newQueue()
  private val cleanupTask =

      Supplier<T> delegate, long duration, TimeUnit unit) {
    checkNotNull(delegate);
    checkArgument(duration > 0, "duration (%s %s) must be > 0", duration, unit);
    return new ExpiringMemoizingSupplier<>(delegate, unit.toNanos(duration));
  }

  /**
   * Returns a supplier that caches the instance supplied by the delegate and removes the cached
   * value after the specified time has passed. Subsequent calls to {@code get()} return the cached

    }

    void sleepMillis(int millis) {
      sleepMicros("U", MILLISECONDS.toMicros(millis));
    }

    void sleepMicros(String caption, long micros) {
      instant += MICROSECONDS.toNanos(micros);
      events.add(caption + String.format(Locale.ROOT, "%3.2f", micros / 1000000.0));
    }

    @Override
    protected void sleepMicrosUninterruptibly(long micros) {
      sleepMicros("R", micros);

    }

    void sleepMillis(int millis) {
      sleepMicros("U", MILLISECONDS.toMicros(millis));
    }

    void sleepMicros(String caption, long micros) {
      instant += MICROSECONDS.toNanos(micros);
      events.add(caption + String.format(Locale.ROOT, "%3.2f", micros / 1000000.0));
    }

    @Override
    protected void sleepMicrosUninterruptibly(long micros) {
      sleepMicros("R", micros);

      Supplier<T> delegate, long duration, TimeUnit unit) {
    checkNotNull(delegate);
    checkArgument(duration > 0, "duration (%s %s) must be > 0", duration, unit);
    return new ExpiringMemoizingSupplier<>(delegate, unit.toNanos(duration));
  }

  /**
   * Returns a supplier that caches the instance supplied by the delegate and removes the cached
   * value after the specified time has passed. Subsequent calls to {@code get()} return the cached

    assertEquals(unit.toNanos(duration), map.expireAfterWriteNanos);
  }

  public void testSetExpireAfterAccess() {
    long duration = 42;
    TimeUnit unit = SECONDS;
    LocalCache<Object, Object> map =
        makeLocalCache(createCacheBuilder().expireAfterAccess(duration, unit));
    assertEquals(unit.toNanos(duration), map.expireAfterAccessNanos);
  }

    assertFailsWith<SocketTimeoutException> {
      call.execute()
    }
    val elapsedNanos = System.nanoTime() - startNanos
    org.junit.jupiter.api.Assertions.assertTrue(
      elapsedNanos < TimeUnit.SECONDS.toNanos(5),
      "Timeout should have taken ~100ms but was " + elapsedNanos / 1e6 + " ms",
    )
  }

  @Test
  fun chainWithReadTimeout() {
    val interceptor1 =
      Interceptor { chainA: Interceptor.Chain ->

    }
  }

  /**
   * Returns unit.toNanos(time), additionally ensuring the returned value is not at risk of
   * overflowing or underflowing, by bounding the value between 0 and (Long.MAX_VALUE / 4) * 3.
   * Actually waiting for more than 219 years is not supported!
   */
  private static long toSafeNanos(long time, TimeUnit unit) {
    long timeoutNanos = unit.toNanos(time);

  // Guarded by this.
  private val currentPushRequests = mutableSetOf<Int>()

  init {
    if (builder.pingIntervalMillis != 0) {
      val pingIntervalNanos = TimeUnit.MILLISECONDS.toNanos(builder.pingIntervalMillis.toLong())
      writerQueue.schedule("$connectionName ping", pingIntervalNanos) {
        val failDueToMissingPong =
          withLock {
            if (intervalPongsReceived < intervalPingsSent) {

        if (effect.closeSocket) socket.close()
      }

      Stall -> {
        // Sleep until the socket is closed.
        socket.sleepWhileOpen(TimeUnit.MINUTES.toNanos(60))
        error("expected timeout")
      }
    }

    return true
  }

  @Throws(IOException::class)
  private fun readEmptyLine(source: BufferedSource) {