Http1ExchangeCodec(client, this, source, sink)
    }
  }

  @Throws(SocketException::class)
  internal fun newWebSocketStreams(exchange: Exchange): RealWebSocket.Streams {
    val socket = this.socket!!
    val source = this.source!!
    val sink = this.sink!!

    socket.soTimeout = 0
    noNewExchanges()
    return object : RealWebSocket.Streams(true, source, sink) {
      override fun close() {

     * reader lock and return the new bytes.
     *
     * ## The file
     *
     * In this case we copy bytes from the file to the [sink].
     *
     * ## The buffer
     *
     * In this case the bytes are immediately copied into [sink] and the number of bytes copied is
     * returned.
     *
     * If upstream would be selected but another thread is already reading upstream this will

   */
  private inner class ChunkedSink : Sink {
    private val timeout = ForwardingTimeout(sink.timeout())
    private var closed: Boolean = false

    override fun timeout(): Timeout = timeout

    override fun write(
      source: Buffer,
      byteCount: Long,
    ) {
      check(!closed) { "closed" }
      if (byteCount == 0L) return

    fun writeTo(editor: DiskLruCache.Editor) {
      editor.newSink(ENTRY_METADATA).buffer().use { sink ->
        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(": ")

      val source = this.source!!
      val sink = this.sink!!
      val tunnelCodec =
        Http1ExchangeCodec(
          // No client for CONNECT tunnels:
          client = null,
          carrier = this,
          source = source,
          sink = sink,
        )
      source.timeout().timeout(readTimeoutMillis.toLong(), TimeUnit.MILLISECONDS)

     *
     * By default this certificate cannot not sign other certificates. Set this to 0 so this
     * certificate can sign other certificates (but those certificates cannot themselves sign
     * certificates). Set this to 1 so this certificate can sign intermediate certificates that can
     * themselves sign certificates. Add one for each additional layer of intermediates to permit.
     */

      object : RequestBody() {
        override fun contentType(): MediaType? {
          return null
        }

        override fun writeTo(sink: BufferedSink) {
          sink.writeUtf8("Hello request!")
          sink.close()
        }

        override fun isDuplex(): Boolean {
          return true
        }
      }
    val request =
      request()

    require(streamId and 0x80000000.toInt() == 0) { "reserved bit set: $streamId" }
    sink.writeMedium(length)
    sink.writeByte(type and 0xff)
    sink.writeByte(flags and 0xff)
    sink.writeInt(streamId and 0x7fffffff)
  }

  @Throws(IOException::class)
  override fun close() {
    this.withLock {
      closed = true
      sink.close()
    }
  }

  @Throws(IOException::class)

        return original!!.contentLength()
      }

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

  private fun uppercase(original: BufferedSink): Sink {
    return object : ForwardingSink(original) {
      override fun write(

    server.listener.assertClosing(1000, "Bye!")
    server.webSocket!!.finishReader()
    server.listener.assertClosed(1000, "Bye!")
  }

  @Test
  fun emptyCloseInitiatesShutdown() {
    server.sink.write("8800".decodeHex()).emit() // Close without code.
    client.processNextFrame()
    client.listener.assertClosing(1005, "")
    client.webSocket!!.finishReader()