notifyAll()
      if (source.finished && sink.finished) {
        return false
      }
    }
    connection.removeStream(id)
    return true
  }

  @Throws(IOException::class)
  fun receiveData(
    source: BufferedSource,
    length: Int,
  ) {
    assertLockNotHeld()

    this.source.receive(source, length.toLong())
  }

        writeSynResetLater(streamId, ErrorCode.PROTOCOL_ERROR)
        updateConnectionFlowControl(length.toLong())
        source.skip(length.toLong())
        return
      }
      dataStream.receiveData(source, length)
      if (inFinished) {
        dataStream.receiveHeaders(Headers.EMPTY, true)
      }
    }

    override fun headers(
      inFinished: Boolean,
      streamId: Int,

          .build(),
      ).assertBody("abc")
      .assertSentRequestAtMillis(sentAt, receivedAt)
      .assertReceivedResponseAtMillis(sentAt, receivedAt)
    val recordedRequest = server.takeRequest()
    assertThat(recordedRequest.method).isEqualTo("GET")
    assertThat(recordedRequest.headers["User-Agent"]).isEqualTo("SyncApiTest")
    assertThat(recordedRequest.body).isNull()