}
  }

  /** A sink that writes outgoing data frames of a stream. This class is not thread safe. */
  inner class FramingSink(
    /** True if either side has cleanly shut down this stream. We shall send no more bytes. */
    var finished: Boolean = false,
  ) : Sink {
    /**
     * Buffer of outgoing data. This batches writes of small writes into this sink as larges frames

          }

          @Override public void writeTo(BufferedSink sink) throws IOException {
            BufferedSink bufferedSink = Okio.buffer(sink(sink));
            delegate.writeTo(bufferedSink);
            bufferedSink.flush();
          }

          public Sink sink(Sink sink) {
            return new ForwardingSink(sink) {
              private long totalBytesWritten = 0L;

  private object SlowRequestBody : RequestBody() {
    override fun contentType(): MediaType? = null

    override fun writeTo(sink: BufferedSink) {
      for (i in 0 until 50) {
        sink.writeUtf8("abc")
        sink.flush()
        Thread.sleep(100)
      }
      fail("")
    }
  }

    }

    private Source source(Source source) {
      return new ForwardingSource(source) {
        long totalBytesRead = 0L;

        @Override public long read(Buffer sink, long byteCount) throws IOException {
          long bytesRead = super.read(sink, byteCount);
          // read() returns the number of bytes read, or -1 if this source is exhausted.
          totalBytesRead += bytesRead != -1 ? bytesRead : 0;

}

internal class BufferMockResponseBody(
  val buffer: Buffer,
) : MockResponseBody {
  override val contentLength = buffer.size

  override fun writeTo(sink: BufferedSink) {
    buffer.copyTo(sink.buffer)
    sink.emitCompleteSegments()
  }

    @Throws(IOException::class)
    override fun contentLength(): Long = delegate.contentLength()

    @Throws(IOException::class)
    override fun writeTo(sink: BufferedSink) {
      val forwardingSink =
        object : ForwardingSink(sink) {
          private var totalBytesWritten: Long = 0
          private var completed = false

          override fun write(
            source: Buffer,

    val buffer = Buffer()

    val message1 = "hello".encodeUtf8()
    val message2 = "hello 2".encodeUtf8()

    DeflaterSink(buffer, Deflater(Deflater.DEFAULT_COMPRESSION, true)).use { sink ->
      sink.write(Buffer().write(message1), message1.size.toLong())
    }
    buffer.writeByte(0x00)
    // Trailing data. We use the Okio segment size to make sure it's still in the input buffer.

      override fun writeTo(sink: BufferedSink) {
        sink.writeUtf8("<el")
        sink.flush()
        throw IOException("failed to stream the XML")
      }
    }

    class ValidRequestBody : RequestBody() {
      override fun contentType(): MediaType = "application/xml".toMediaType()

      override fun writeTo(sink: BufferedSink) {
        sink.writeUtf8("<element/>")
        sink.flush()

    get() = stream?.isSourceComplete == true

  override val socket: Socket
    get() = stream!!

  override fun createRequestBody(
    request: Request,
    contentLength: Long,
  ): Sink = stream!!.sink

  override fun writeRequestHeaders(request: Request) {
    if (stream != null) return

    val hasRequestBody = request.body != null
    val requestHeaders = http2HeadersList(request)

      override fun writeTo(sink: BufferedSink) {
        val uppercase = uppercase(sink)
        val bufferedSink = uppercase.buffer()
        original!!.writeTo(bufferedSink)
        bufferedSink.emit()
      }
    }

  private fun uppercase(original: BufferedSink): Sink =
    object : ForwardingSink(original) {
      override fun write(
        source: Buffer,