Ktor 响应流答案

【问题标题】：Ktor response streamingKtor 响应流
【发布时间】：2021-02-26 01:05:54
【问题描述】：

我正在尝试调用一个 twitter 端点，它会为您提供源源不断的 json 结果流并返回给客户端

https://documenter.getpostman.com/view/9956214/T1LMiT5U#977c147d-0462-4553-adfa-d7a1fe59c3ec

我尝试像这样调用端点

        val url = "https://api.twitter.com/2/tweets/search/stream"
        _streamChannel = _client.get<ByteReadChannel>(token, url) //Stops here

        val byteBufferSize = 1024
        val byteBuffer = ByteArray(byteBufferSize)

        _streamChannel?.let {
            while (_streamChannel!!.availableForRead > 0) {
                _streamChannel!!.readAvailable(byteBuffer, 0, byteBufferSize)
                val s = String(byteBuffer)
                parseStreamResponseString(s).forEach {
                    emit(Response.Success(it))
                }
            }
        }

我的client.get 代码是这个

suspend inline fun <reified T> get(authKey: String, url: String): T? {
    val response = _client.get<HttpResponse>(url) {
        header("Authorization", "Bearer $authKey")
    }

    when (response.status.value) {
        in 300..399 -> throw RedirectResponseException(response)
        in 400..499 -> throw ClientRequestException(response)
        in 500..599 -> throw ServerResponseException(response)
    }

    if (response.status.value >= 600) {
        throw ResponseException(response)
    }

    return response.receive<T>()
}

当我提出请求时，它只是坐在那里，我假设是等待返回完整的响应，然后再将其提供给我。

编辑

我也尝试使用scoped streaming，但它只是位于readAvailable 行我知道有消息通过，因为当我通过cURL 运行请求时，我不断获取数据

    _client.get<HttpStatement> {
        header("Authorization", "Bearer $authKey")
        url(urlString)
        contentType(ContentType.Application.Json)
        method = HttpMethod.Get
    }.execute {
        val streamChannel = it.receive<ByteReadChannel>()
        val byteBufferSize = 1024
        val byteBuffer = ByteArray(byteBufferSize)
        streamChannel.readAvailable(byteBuffer, 0, byteBufferSize) // Stops here
        val s = String(byteBuffer)
    }

如何使用 Ktor 处理恒定的 json 数据流？

【问题讨论】：

标签： ktor http-streaming

【解决方案1】：

据我所知，Ktor 客户端确实以 twitter 的流 API 所需的方式公开了对请求的 IO 缓冲区的访问。

来自推特文档here：

某些 HTTP 客户端库仅在服务器关闭连接后才返回响应正文。这些客户端将无法访问 Streaming API。您必须使用将逐步返回响应数据的 HTTP 客户端。大多数健壮的 HTTP 客户端库将提供此功能。例如，Apache HttpClient 将处理这个用例。

你正在做的是告诉 Ktor 你得到的是一个ByteReadChannel，因此，一旦请求关闭（这个 twitter 端点永远不会发生），Ktor 客户端将尝试使用任何插件（json例如）您用来将该数据解析为ByteReadChannel。它也无法做到这一点，因为您从 twitter 获得的数据不是 ByteReadChannel，它是一个新行分隔的 json 对象列表。

【讨论】：