}

        // Wait for an in-flight connect to complete or fail.
        if (connectResult == null) {
          connectResult = awaitTcpConnect(awaitTimeoutNanos, TimeUnit.NANOSECONDS) ?: continue
        }

        if (connectResult.isSuccess) {
          // We have a connected TCP connection. Cancel and defer the racing connects that all lost.
          cancelInFlightConnects()

  public static RateLimiter create(double permitsPerSecond, Duration warmupPeriod) {
    return create(permitsPerSecond, toNanosSaturated(warmupPeriod), TimeUnit.NANOSECONDS);
  }

  /**
   * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per
   * second" (commonly referred to as <i>QPS</i>, queries per second), and a <i>warmup period</i>,

    val startNanos = System.nanoTime()
    while (webSocket.receivedPongCount() < 3) {
      Thread.sleep(50)
    }
    val elapsedUntilPong3 = System.nanoTime() - startNanos
    assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedUntilPong3).toDouble())
      .isCloseTo(1500.0, 250.0)

    // The client pinged the server 3 times, and it has ponged back 3 times.
    assertThat(webSocket.sentPingCount()).isEqualTo(3)

   */
  public static RateLimiter create(double permitsPerSecond, Duration warmupPeriod) {
    return create(permitsPerSecond, toNanosSaturated(warmupPeriod), TimeUnit.NANOSECONDS);
  }

  /**
   * Creates a {@code RateLimiter} with the specified stable throughput, given as "permits per
   * second" (commonly referred to as <i>QPS</i>, queries per second), and a <i>warmup period</i>,

import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;

import com.google.common.base.Equivalence;
import com.google.common.base.Ticker;
import com.google.common.cache.LocalCache.EntryFactory;

import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;

import com.google.common.base.Equivalence;
import com.google.common.base.Ticker;
import com.google.common.cache.LocalCache.EntryFactory;

    assertFailsWith<InterruptedIOException> {
      stream.takeHeaders()
    }
    val elapsedNanos = System.nanoTime() - startNanos
    awaitWatchdogIdle()
    // 200ms delta
    assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedNanos).toDouble())
      .isCloseTo(500.0, 200.0)
    assertThat(connection.openStreamCount()).isEqualTo(0)

    // Verify the peer received what was expected.

      // a test timeout failure.
      val waitTime = (entryTime + 1_000_000_000L - System.nanoTime())
      if (!queue.idleLatch().await(waitTime, TimeUnit.NANOSECONDS)) {
        TaskRunner.INSTANCE.lock.withLock {
          TaskRunner.INSTANCE.cancelAll()
        }
        fail<Unit>("Queue still active after 1000 ms")
      }
    }
  }

    lastTimestampNs = result.timestampNs

    if (eventClass != null) {
      assertThat(result).isInstanceOf(eventClass)
    }

    if (elapsedMs != -1L) {
      assertThat(
        TimeUnit.NANOSECONDS.toMillis(actualElapsedNs)
          .toDouble(),
      )
        .isCloseTo(elapsedMs.toDouble(), 100.0)
    }

    return result
  }

  fun recordedEventTypes() = eventSequence.map { it.name }

    val timeoutBefore = source.timeout().timeoutNanos()
    source.timeout().clearTimeout()
    try {
      b0 = source.readByte() and 0xff
    } finally {
      source.timeout().timeout(timeoutBefore, TimeUnit.NANOSECONDS)
    }

    opcode = b0 and B0_MASK_OPCODE
    isFinalFrame = b0 and B0_FLAG_FIN != 0
    isControlFrame = b0 and OPCODE_FLAG_CONTROL != 0