e: IOException?,
  ) {
    var noNewExchangesEvent = false
    lock.withLock {
      if (e is StreamResetException) {
        when {
          e.errorCode == ErrorCode.REFUSED_STREAM -> {
            // Stop using this connection on the 2nd REFUSED_STREAM error.
            refusedStreamCount++
            if (refusedStreamCount > 1) {
              noNewExchangesEvent = !noNewExchanges
              noNewExchanges = true

            minDelayNanos = minOf(candidateDelay, minDelayNanos)
            continue@eachQueue
          }

          // If we already have more than one task, that's enough work for now. Stop searching.
          readyTask != null -> {
            multipleReadyTasks = true
            break@eachQueue
          }

          // We have a task to execute when we complete the loop.
          else -> {

        flags = if (ack) FLAG_ACK else FLAG_NONE,
      )
      sink.writeInt(payload1)
      sink.writeInt(payload2)
      sink.flush()
    }
  }

  /**
   * Tell the peer to stop creating streams and that we last processed `lastGoodStreamId`, or zero
   * if no streams were processed.
   *
   * @param lastGoodStreamId the last stream ID processed, or zero if no streams were processed.

    val e = InterruptedIOException("timeout")
    if (cause != null) e.initCause(cause)
    @Suppress("UNCHECKED_CAST") // E is either IOException or IOException?
    return e as E
  }

  /**
   * Stops applying the timeout before the call is entirely complete. This is used for WebSockets
   * and duplex calls where the timeout only applies to the initial setup.
   */
  fun timeoutEarlyExit() {
    check(!timeoutEarlyExit)

  /**
   * Degrades this connection such that new streams can neither be created locally, nor accepted
   * from the remote peer. Existing streams are not impacted. This is intended to permit an endpoint
   * to gracefully stop accepting new requests without harming previously established streams.
   */
  @Throws(IOException::class)
  fun shutdown(statusCode: ErrorCode) {
    writer.withLock {
      val lastGoodStreamId: Int

 *    we must send a `CONNECT` request, and handle authorization challenges from the proxy.
 *  * Optional [TLS handshake][connectTls].
 *
 * Each step may fail. If a retry is possible, a new instance is created with the next plan, which
 * will be configured differently.
 */
class ConnectPlan(
  private val taskRunner: TaskRunner,

 * the same structure:
 *
 *  * prepare a set of plans, each with a predictable connect delay
 *  * attempt to find a connection
 *  * step through time, asserting that the expected side effects are performed.
 */
internal class FastFallbackExchangeFinderTest {
  private val taskFaker = TaskFaker()
  private val taskRunner = taskFaker.taskRunner

  /**