return ChunkedSource(url)
  }

  private fun newUnknownLengthSource(url: HttpUrl): Source {
    check(state == STATE_OPEN_RESPONSE_BODY) { "state: $state" }
    state = STATE_READING_RESPONSE_BODY
    carrier.noNewExchanges()
    return UnknownLengthSource(url)
  }

  /**
   * Sets the delegate of `timeout` to [Timeout.NONE] and resets its underlying timeout

    assertThat(server2.requestCount).isEqualTo(1)

    assertThat(clientTestRule.recordedConnectionEventTypes()).containsExactly(
      "ConnectStart",
      "ConnectEnd",
      "ConnectionAcquired",
      "NoNewExchanges",
      "ConnectionReleased",
      "ConnectionClosed",
      "ConnectStart",
      "ConnectEnd",
      "ConnectionAcquired",
      "ConnectionReleased",
    )
  }

  @Test

            // Everything else wants a fresh connection.
            noNewExchangesEvent = !noNewExchanges
            noNewExchanges = true
            routeFailureCount++
          }
        }
      } else if (!isMultiplexed || e is ConnectionShutdownException) {
        noNewExchangesEvent = !noNewExchanges
        noNewExchanges = true

        // If this route hasn't completed a call, avoid it for new connections.

   */
  interface Carrier {
    val route: Route

    fun trackFailure(
      call: RealCall,
      e: IOException?,
    )

    fun noNewExchanges()

    fun cancel()
  }

  companion object {
    /**
     * The timeout to use while discarding a stream of input data. Since this is used for connection

    }
    assertThat(listener.recordedEventTypes()).containsExactly(
      "ConnectStart",
      "ConnectEnd",
      "ConnectionAcquired",
      "NoNewExchanges",
      "ConnectionReleased",
      "ConnectionClosed",
    )
  }

  @Throws(IOException::class)
  private fun assertSuccessfulEventOrder() {
    val call =
      client.newCall(

    return connection
  }

  override fun trackFailure(
    call: RealCall,
    e: IOException?,
  ) {
    // Do nothing.
  }

  override fun noNewExchanges() {
    // Do nothing.
  }

  override fun cancel() {
    canceled = true
    // Close the raw socket so we don't end up doing synchronous I/O.
    rawSocket?.closeQuietly()
  }

      override val sink = RequestBodySink(codec.socket.sink, -1L)
      override val source = ResponseBodySource(codec.socket.source, -1L)
    }
  }

  fun noNewExchangesOnConnection() {
    codec.carrier.noNewExchanges()
  }

  fun cancel() {
    codec.cancel()
  }

  /**
   * Revoke this exchange's access to streams. This is necessary when a follow-up request is

    awaitGarbageCollection()
    assertThat(pool.closeConnections(100L)).isEqualTo(0L)
    assertThat(c1.calls).isEmpty()

    // Can't allocate once a leak has been detected.
    assertThat(c1.noNewExchanges).isTrue()
  }

  @Test fun interruptStopsThread() {
    val taskRunnerThreads = mutableListOf<Thread>()
    val taskRunner =
      TaskRunner(
        RealBackend { runnable ->