title: Decode chunked JSON with AKKA Stream
body: <p>I have a <code>Source[ByteString, _]</code> from an input file with 3 rows like this (in reality the input is a TCP socket with a continuos stream):</p>
  
  <pre><code>{"a":[2
  33]
  }
  </code></pre>
  
  <p>Now the problem is that I want to parse this into a <code>Source[ChangeMessage,_]</code>, however the only examples I have found deals with when there is a whole JSON message for every row not when each JSON message can be fragmented over multiple rows.</p>
  
  <p>One example I found is this <a href="https://github.com/knutwalker/akka-stream-json" rel="noreferrer">this</a> library, however it expects <code>}</code> or <code>,</code> as last character, that is one JSON per row. The example below shows this setup.</p>
  
  <pre><code>"My decoder" should "decode chunked json" in {
      implicit val sys = ActorSystem("test")
      implicit val mat = ActorMaterializer()
      val file = Paths.get("chunked_json_stream.json")
      val data = FileIO.fromPath(file)
      .via(CirceStreamSupport.decode[ChangeMessage])
      .runWith(TestSink.probe[ChangeMessage])
      .request(1)
      .expectComplete()
    }
  </code></pre>
  
  <p>Another alternative would be to use a fold and balance <code>}</code> and only emit when a whole JSON is completed. The problem with this is that the fold operator only emits when the stream completes and since this is a continuous stream I can not use it here.</p>
  
  <blockquote>
    <p>My question is: What is the fastest way to parse chunked JSON streams
    in AKKA Stream and are there any available software that already does
    this? If possible I would like to use <a href="https://circe.github.io/circe/" rel="noreferrer">circe</a></p>
  </blockquote>