private fun transfer(
    fromAddress: InetAddress,
    toAddress: InetAddress,
    source: BufferedSource,
    sink: BufferedSink,
  ) {
    executor.execute {
      val name = "SocksProxy $fromAddress to $toAddress"
      threadName(name) {
        val buffer = Buffer()
        try {
          sink.use {
            source.use {
              while (true) {
                val byteCount = source.read(buffer, 8192L)

    return object : ForwardingSource(source) {
      @Throws(IOException::class)
      override fun read(
        sink: Buffer,
        byteCount: Long,
      ): Long {
        val buffer = Buffer()
        val read = delegate.read(buffer, byteCount)
        if (read != -1L) {
          sink.write(buffer.readByteString().toAsciiUppercase())
        }
        return read
      }
    }
  }

          .post(
            object : RequestBody() {
              override fun contentType() = null

              override fun writeTo(sink: BufferedSink) {
                for (i in 0..9) {
                  sink.writeByte(0)
                  sink.flush()
                  sleep(100)
                }
                fail("Expected connection to be closed")
              }
            },

      mutableByteCount -= bytesWritten
    }
  }

  /**
   * Copy [byteCount] bytes from the file at [pos] into `sink`. It is the
   * caller's responsibility to make sure there are sufficient bytes to read: if there aren't this
   * method throws an `EOFException`.
   */
  fun read(
    pos: Long,
    sink: Buffer,
    byteCount: Long,
  ) {
    if (byteCount < 0L) {
      throw IndexOutOfBoundsException()
    }

    val pipe = Pipe(1024)
    val sink = pipe.sink.buffer()
    val relay = edit(file, pipe.source, metadata, 5)
    val future1 = executor.submit(sourceReader(relay.newSource()))
    val future2 = executor.submit(sourceReader(relay.newSource()))
    Thread.sleep(500)
    sink.writeUtf8("abcdefghij")
    Thread.sleep(500)
    sink.writeUtf8("klmnopqrst")
    sink.close()
    assertThat<ByteString>(future1.get())

    val stream = connection.newStream(headerEntries("b", "banana"), true)
    val sink = stream.sink
    sink.write(Buffer().writeUtf8("abcde"), 5)
    stream.writeTimeout().timeout(500, TimeUnit.MILLISECONDS)
    val startNanos = System.nanoTime()
    sink.write(Buffer().writeUtf8("f"), 1)
    assertFailsWith<InterruptedIOException> {
      sink.flush() // This will time out waiting on the write window.
    }

    )
    val asyncRequestBody: RequestBody =
      object : RequestBody() {
        override fun contentType(): MediaType? = null

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

        override fun isDuplex(): Boolean = true
      }
    val request =
      request()
        .post(asyncRequestBody)

      override fun contentType(): MediaType? = "text/plain".toMediaTypeOrNull()

      @Throws(IOException::class)
      override fun writeTo(sink: BufferedSink) {
        try {
          sink.writeUtf8("abc")
          sink.flush()
          Thread.sleep(sleepMillis.toLong())
          sink.writeUtf8("def")
        } catch (e: InterruptedException) {
          throw InterruptedIOException()
        }
      }
    }

    s: String,
    responseSent: CountDownLatch = CountDownLatch(1),
  ) = apply {
    actions += { stream ->
      stream.sink.writeUtf8(s)
      stream.sink.flush()
      responseSent.countDown()
    }
  }

  fun exhaustResponse() =
    apply {
      actions += { stream -> stream.sink.close() }
    }

  fun sleep(
    duration: Long,
    unit: TimeUnit,
  ) = apply {

        var closed = false

        override fun close() {
          closed = true
        }

        override fun read(
          sink: Buffer,
          byteCount: Long,
        ): Long {
          check(!closed)
          return data.read(sink, byteCount)
        }

        override fun timeout(): Timeout = Timeout.NONE
      }
    return source.buffer().asResponseBody(null, -1)
  }