scala - akka-http 分块响应连接

Question

我正在使用 akka-http向发送回分块响应的 http 服务发出请求。这是相关代码的样子:

val httpRequest: HttpRequest = //build the request
val request = Http().singleRequest(httpRequest)
request.flatMap { response =>
    response.entity.dataBytes.runForeach { chunk =>
        println("-----")
        println(chunk.utf8String)
    }
}

命令行中生成的输出如下所示:

-----
{"data":
-----
"some text"}

-----
{"data":
-----
"this is a longer
-----
text"}

-----
{"data": "txt"}

-----
...

数据的逻辑部分 - 在本例中为 json 以行尾符号 \r\n 结尾，但问题是 json 并不总是适合单个 http 响应块，如上例所示。

我的问题是 - 我如何将传入的分块数据连接成完整的 json，以便生成的容器类型仍将保持为 Source[Out,M1]或 Flow[In,Out,M2] ?我想遵循 akka-stream的理念.

更新:还值得一提的是，响应是无止境的，必须实时进行聚合

Answer 1

找到了解决办法:

val request: HttpRequest = //build the request
request.flatMap { response =>
    response.entity.dataBytes.scan("")((acc, curr) => if (acc.contains("\r\n")) curr.utf8String else acc + curr.utf8String)
        .filter(_.contains("\r\n"))
        .runForeach { json =>
            println("-----")
            println(json)
        }
}