akka-http chunked响应连接

Question

我正在使用akka-http一个请求发送回chunked响应的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响应块,如上例中清晰可见.

我的问题是 - 如何将传入的chunked数据连接成完整的jsons,以便生成的容器类型仍然保留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)
        }
}