我正在尝试调用一个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>()
}当我提出请求时,它就坐在那里,我假设它是在等待完整的回复,然后再给我。
编辑
我也尝试过使用作用域流,但是它位于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数据流?
发布于 2021-09-26 05:47:40
据我所知,Ktor客户端确实以twitter的流API所需的方式公开了对请求的IO缓冲区的访问。
来自twitter文档这里
一些HTTP客户端库只在连接被服务器关闭后返回响应体。这些客户端将无法访问流API。您必须使用一个HTTP客户端,该客户端将递增地返回响应数据。大多数健壮的HTTP客户端库将提供此功能。例如,Apache HttpClient将处理这个用例。
您要做的是告诉Ktor,您得到的是一个ByteReadChannel,因此,一旦请求关闭(这个Twitter端点永远不会发生这种情况),Ktor客户端将尝试使用您使用的任何插件(例如json)将数据解析为一个ByteReadChannel。它也无法做到这一点,因为您从twitter获得的数据不是ByteReadChannel,而是一个单独列出json对象的新行。
https://stackoverflow.com/questions/64830813
复制相似问题