type countUpReader struct {
	reader    io.Reader
	bytesRead int64
}

// Max bzip2 concurrency across calls. 50% of GOMAXPROCS.
var bz2Limiter = pbzip2.CreateConcurrencyPool((runtime.GOMAXPROCS(0) + 1) / 2)

func (r *countUpReader) Read(p []byte) (n int, err error) {
	n, err = r.reader.Read(p)
	atomic.AddInt64(&r.bytesRead, int64(n))
	return n, err
}

func (r *countUpReader) BytesRead() int64 {
	if r == nil {
		return 0
	}

            }
            throw e
          }

          if (bytesRead == -1L) {
            if (!cacheRequestClosed) {
              cacheRequestClosed = true
              cacheBody.close() // The cache response is complete!
            }
            return -1
          }

          sink.copyTo(cacheBody.buffer, sink.size - bytesRead, bytesRead)
          cacheBody.emitCompleteSegments()
          return bytesRead

    }
    var mutablePos = pos
    var mutableByteCount = byteCount

    while (mutableByteCount > 0L) {
      val bytesRead = fileChannel.transferTo(mutablePos, mutableByteCount, sink)
      mutablePos += bytesRead
      mutableByteCount -= bytesRead
    }
  }

  }

  fun <E : IOException?> bodyComplete(
    bytesRead: Long,
    responseDone: Boolean,
    requestDone: Boolean,
    e: E,
  ): E {
    if (e != null) {
      trackFailure(e)
    }
    if (requestDone) {
      if (e != null) {
        eventListener.requestFailed(call, e)
      } else {
        eventListener.requestBodyEnd(call, bytesRead)
      }
    }
    if (responseDone) {

	n, err := r.src.Read(p)
	r.bytesRead += int64(n)
	if r.sha256 != nil {
		r.sha256.Write(p[:n])
	}
	if r.contentHasher != nil {
		r.contentHasher.Write(p[:n])
	}

	if err == io.EOF { // Verify content SHA256, if set.
		if r.expectedMin > 0 {
			if r.bytesRead < r.expectedMin {
				return 0, SizeTooSmall{Want: r.expectedMin, Got: r.bytesRead}
			}
		}
		if r.expectedMax > 0 {

    val totalBytesToRead = inflater.bytesRead + deflatedBytes.size

    // We cannot read all, as the source does not close.
    // Instead, we ensure that all bytes from source have been processed by inflater.
    do {
      inflaterSource.readOrInflate(buffer, Long.MAX_VALUE)
    } while (inflater.bytesRead < totalBytesToRead && !inflater.finished())
  }

  @Throws(IOException::class)

          commit(upstreamPos)
          return -1L
        }

        // Update this source and prepare this call's result.
        val bytesRead = minOf(upstreamBytesRead, byteCount)
        upstreamBuffer.copyTo(sink, 0, bytesRead)
        sourcePos += bytesRead

        // Append the upstream bytes to the file.
        fileOperator!!.write(
          FILE_HEADER_SIZE + upstreamPos,

  private class CountingSource(source: Source) : ForwardingSource(source) {
    var bytesRead = 0L

    override fun read(
      sink: Buffer,
      byteCount: Long,
    ): Long {
      val result = delegate.read(sink, byteCount)
      if (result == -1L) return -1L
      bytesRead += result
      return result
    }
  }

  }

  /**
   * We had a bug where self-finishing inflater streams would infinite loop!
   * https://github.com/square/okhttp/issues/8078
   */
  @Test fun `inflate returns finished before bytesRead reaches input length`() {
    val inflater = MessageInflater(false)
    val message = "53621260020000".decodeHex()
    assertThat(inflater.inflate(message)).isEqualTo("22021002".decodeHex())
  }

    override val timestampNs: Long,
    override val call: Call,
  ) : CallEvent()

  data class ResponseBodyEnd(
    override val timestampNs: Long,
    override val call: Call,
    val bytesRead: Long,
  ) : CallEvent() {
    override fun closes(event: CallEvent): Boolean = event is ResponseBodyStart && call == event.call
  }

  data class ResponseFailed(
    override val timestampNs: Long,