) {
    check(state == STATE_IDLE) { "state: $state" }
    socket.sink.writeUtf8(requestLine).writeUtf8("\r\n")
    for (i in 0 until headers.size) {
      socket.sink
        .writeUtf8(headers.name(i))
        .writeUtf8(": ")
        .writeUtf8(headers.value(i))
        .writeUtf8("\r\n")
    }
    socket.sink.writeUtf8("\r\n")
    state = STATE_OPEN_REQUEST_BODY
  }

        sink.writeUtf8(url.toString()).writeByte('\n'.code)
        sink.writeUtf8(requestMethod).writeByte('\n'.code)
        sink.writeDecimalLong(varyHeaders.size.toLong()).writeByte('\n'.code)
        for (i in 0 until varyHeaders.size) {
          sink
            .writeUtf8(varyHeaders.name(i))
            .writeUtf8(": ")
            .writeUtf8(varyHeaders.value(i))
            .writeByte('\n'.code)

      requestBody.writeUtf8("request A\n")
      requestBody.flush()
      val responseBody = response.body.source()
      assertThat(responseBody.readUtf8Line())
        .isEqualTo("response B")
      requestBody.writeUtf8("request C\n")
      requestBody.flush()
      assertThat(responseBody.readUtf8Line())
        .isEqualTo("response D")
      requestBody.writeUtf8("request E\n")
      requestBody.flush()

        socket.sink.buffer().use { sink ->
          socket.source.buffer().use { source ->
            sink.writeUtf8("client says hello\n")
            sink.flush()

            assertThat(source.readUtf8Line()).isEqualTo("server says hello")

            sink.writeUtf8("client says goodbye\n")
            sink.flush()

            assertThat(source.readUtf8Line()).isEqualTo("server says goodbye")

    val sink1 = stream1.sink.buffer()
    val sink2 = stream2.sink.buffer()
    sink1.writeUtf8("abc")
    assertFailsWith<IOException> {
      sink2.writeUtf8("abc")
      sink2.flush()
    }.also { expected ->
      assertThat(expected.message).isEqualTo("stream was reset: REFUSED_STREAM")
    }
    sink1.writeUtf8("def")
    sink1.close()
    assertFailsWith<ConnectionShutdownException> {

      )
    stream.writeHeaders(
      responseHeaders = responseHeaders,
      outFinished = false,
      flushHeaders = false,
    )
    val out = stream.sink.buffer()
    out.writeUtf8("Not found: $path")
    out.close()
  }

  private fun serveDirectory(
    stream: Http2Stream,
    files: Array<File>,
  ) {
    val responseHeaders =
      listOf(

          val chunkSize = minOf(body.size, maxChunkSize.toLong())
          bytesOut.writeHexadecimalUnsignedLong(chunkSize)
          bytesOut.writeUtf8("\r\n")
          bytesOut.write(body, chunkSize)
          bytesOut.writeUtf8("\r\n")
        }
        bytesOut.writeUtf8("0\r\n") // Last chunk. Trailers follow!
        this.body = bytesOut.toMockResponseBody()
      }

    /**

   * most [BufferedSource] instances, but not of [Buffer].
   */
  private fun responseBody(content: String): ResponseBody {
    val data = Buffer().writeUtf8(content)
    val source: Source =
      object : Source {
        var closed = false

        override fun close() {
          closed = true
        }

        override fun read(
          sink: Buffer,

    assertThat(response.body.string()).isEqualTo("Hello")
  }

  private fun gzip(data: String): Buffer {
    val result = Buffer()
    val sink = GzipSink(result).buffer()
    sink.writeUtf8(data)
    sink.close()
    return result
  }

  private fun writeHeaders(
    sink: BufferedSink,
    headers: Headers,
  ) {
    for ((name, value) in headers) {
      sink.writeUtf8(name)
      sink.writeUtf8(": ")
      sink.writeUtf8(value)
      sink.writeUtf8("\r\n")
    }
    sink.writeUtf8("\r\n")
    sink.flush()
  }

  /** Returns a sink that applies throttling and disconnecting. */
  private fun Sink.withThrottlingAndSocketEffect(