* reached. Returns the total number of bytes read. Does not close the stream.
   *
   * @since 20.0
   */
  @CanIgnoreReturnValue
  public static long exhaust(InputStream in) throws IOException {
    long total = 0;
    long read;
    byte[] buf = createBuffer();
    while ((read = in.read(buf)) != -1) {
      total += read;
    }
    return total;
  }

        .url(server.url("/"))
        .method(method, null)
        .build()
    val response = client.newCall(request).execute()
    assertThat(response.code).isEqualTo(expectedResponseCode)

    // Exhaust the content stream.
    response.body.string()
    val cached = cacheGet(cache, request)
    if (shouldWriteToCache) {
      assertThat(cached).isNotNull()
      cached!!.body.close()
    } else {

  @LazyInit
  @GuardedBy("queue")
  private WorkerRunningState workerRunningState = IDLE;

  /**
   * This counter prevents an ABA issue where a thread may successfully schedule the worker, the
   * worker runs and exhausts the queue, another thread enqueues a task and fails to schedule the
   * worker, and then the first thread's call to delegate.execute() returns. Without this counter,

  @LazyInit
  @GuardedBy("queue")
  private WorkerRunningState workerRunningState = IDLE;

  /**
   * This counter prevents an ABA issue where a thread may successfully schedule the worker, the
   * worker runs and exhausts the queue, another thread enqueues a task and fails to schedule the
   * worker, and then the first thread's call to delegate.execute() returns. Without this counter,

    if (peek == null) {
      if (!exhausted()) return // Expected a token; got something else.
      result.add(Challenge(schemeName, emptyMap()))
      return
    }

    var eqCount = skipAll('='.code.toByte())
    val commaSuffixed = skipCommasAndWhitespace()

    // It's a token68 because there isn't a value after it.
    if (!commaPrefixed && (commaSuffixed || exhausted())) {
      result.add(
        Challenge(

    apply {
      actions += { stream ->
        if (!stream.source.exhausted()) throw AssertionError("expected exhausted")
      }
    }

  fun cancelStream() =
    apply {
      actions += { stream -> stream.cancel() }
    }

  fun requestIOException() =
    apply {
      actions += { stream ->
        try {
          stream.source.exhausted()
          throw AssertionError("expected IOException")

    while (hasNextProxy()) {
      // Postponed routes are always tried last. For example, if we have 2 proxies and all the
      // routes for proxy1 should be postponed, we'll move to proxy2. Only after we've exhausted
      // all the good routes will we attempt the postponed routes.
      val proxy = nextProxy()
      for (inetSocketAddress in inetSocketAddresses) {
        val route = Route(address, proxy, inetSocketAddress)

   * Returns the trailers if they're immediately available.
   */
  @Throws(IOException::class)
  fun peekTrailers(): Headers? {
    withLock {
      if (source.finished && source.receiveBuffer.exhausted() && source.readBuffer.exhausted()) {
        return source.trailers ?: Headers.EMPTY
      }
      if (errorCode != null) {
        throw errorException ?: StreamResetException(errorCode!!)
      }
      return null
    }

    val v1 = utf8.readDecimalLong()
    require(utf8.readByte() == '.'.code.toByte())
    val v2 = utf8.readDecimalLong()
    writeVariableLengthLong(v1 * 40 + v2)

    while (!utf8.exhausted()) {
      require(utf8.readByte() == '.'.code.toByte())
      val vN = utf8.readDecimalLong()
      writeVariableLengthLong(vN)
    }
  }

  fun writeRelativeObjectIdentifier(s: String) {

    get() = if (limit == -1L) -1L else (limit - byteCount)

  fun hasNext(): Boolean = peekHeader() != null

  /**
   * Returns the next header to process unless this scope is exhausted.
   *
   * This returns null if:
   *
   *  * The stream is exhausted.
   *  * We've read all of the bytes of an object whose length is known.
   *  * We've reached the [DerHeader.TAG_END_OF_CONTENTS] of an object whose length is unknown.
   */