akka-httpチャンクレス応答連結

Question

私はakka-httpを使用して、チャンク応答を返すhttpサービスにリクエストします。この場合にはJSON終了 -

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

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

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

----- 
... 

データの論理的な作品：

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

と、コマンドラインで生成される出力は次のようになります。これは、コードの該当ビットがどのように見えるかです行末に\r\nというシンボルが付いていますが、上記の例ではjsonが単一のhttp応答チャンクに収まるとは限りません。

私の質問は、結果のコンテナタイプがまだSource[Out,M1]またはFlow[In,Out,M2]のままになるように、着信チャンクデータを完全なjsonsに連結するにはどうすればよいですか？私はakka-streamの理想に従いたいと思います。

UPDATE：

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) 
     } 
} 

Answer 1

は、ソリューションが見つかりリアルタイムに行われなければならないこと、また、言及する価値があります非常に問題： "ByteStringのストリームから行を解析する"。彼らの解決策は非常に冗長ですが、単一のチャンクが複数の行を含む状況にも対応できます。これは、チャンクサイズが複数のjsonメッセージを処理するのに十分な大きさに変化する可能性があるため、より堅牢に見えます。

Answer 2

akka stream documentationこのため料理のエントリを持っている：それは応答が無限であると凝集が

Answer 3

response.entity.dataBytes 
.via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 8096)) 
.mapAsyncUnordered(Runtime.getRuntime.availableProcessors()) { data => 
    if (response.status == OK) { 
    val event: Future[Event] = Unmarshal(data).to[Event] 
    event.foreach(x => log.debug("Received event: {}.", x)) 
    event.map(Right(_)) 
    } else { 
    Future.successful(data.utf8String) 
     .map(Left(_)) 
    } 
} 

唯一の要件は、1つのレコードの最大サイズを知っていることです。小さいものから始めた場合、デフォルトの動作は、レコードが制限よりも大きい場合に失敗することです。失敗の代わりに切り捨てるように設定できますが、JSONの部分は意味をなさない。