AkkaStreamにはJsonオブジェクトの配列をパースするときにオブジェクトごとに区切ってくれる機能がある。これがJsonFramingである。
たとえば、以下のようなJSON文字列があったとする
[
{"hoge": 1},
{"mogu": 2},
{"goro": 3}
]
３要素のある配列なので、３つの要素に分解して、ストリームが流れてくる。
{"hoge": 1}{"mogu": 2}{"goro": 3}
配列を全部呼んでからパースするのではなく、配列内のオブジェクトを一つ一つパースできるので効率的に使える。
使ってみる
implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
import scala.concurrent.ExecutionContext.Implicits.global
val jsonFraming: Flow[ByteString, ByteString, NotUsed] = JsonFraming.objectScanner(1000) // 今回試す JsonFraming 引数の1000はオブジェクトの最大長
Source.single(ByteString("[{\"hoge\": 1}, {\"goro\": 2}, {\"mogu\": 3}"))
.via(jsonFraming)
.runForeach(n => println(n.decodeString("UTF-8")))
.foreach(_ => system.terminate())
JsonFraming.objectScanner(1000)でオブジェクトの最大長が1000のFlowが作成できる。オブジェクトの長さが1000超えるものがあると停止してしまう。
出力
{"hoge": 1}
{"mogu": 2}
{"goro": 3}