Fully asynchronous http client call with json streamed response using akka-http will work in all cases, even with chunked responses ! / published by https://github.com/dacr/code-examples-manager #25d85f17-5e12-4a74-ba8a-6aa629575af5/c85d5daddb95a353302c521546214e4a1c85516a

akka-http-client-json-stream.sc

// summary : Fully asynchronous http client call with json streamed response using akka-http will work in all cases, even with chunked responses !
// keywords : scala, actors, akka, http-client, client, json, json4s, http-stream, stream
// publish : gist
// authors : David Crosson
// license : Apache License Version 2.0 (https://www.apache.org/licenses/LICENSE-2.0.txt)
// id : 25d85f17-5e12-4a74-ba8a-6aa629575af5
// created-on : 2018-06-19T19:10:04Z
// managed-by : https://github.com/dacr/code-examples-manager
// run-with : scala-cli $file

// ---------------------
//> using scala "2.13.14"
//> using dep "com.typesafe.akka::akka-http:10.2.7"
//> using dep "com.typesafe.akka::akka-stream:2.6.18"
//> using dep "de.heikoseeberger::akka-http-json4s:1.38.2"
//> using dep "org.json4s::json4s-jackson:4.0.3"
//// using lib "org.slf4j:slf4j-nop:1.7.32"
// ---------------------

/*
  akka-http http client request with streamed json responses
  ---
  scala in script mode using ammonite REPL (http://ammonite.io/)
  ---
  take care this example is using the `toStrict(timeout)` method
  to eagerly consume the entity and make it available in memory
*/

import akka.http.scaladsl._
import akka.http.scaladsl.model._
import akka.stream.scaladsl.Framing
import akka.util.ByteString

import scala.concurrent._
import scala.concurrent.duration._
import org.json4s.{DefaultFormats, JValue}
import org.json4s.jackson.JsonMethods._
import org.json4s._


object TestThat {
  implicit val system = akka.actor.ActorSystem("MySystem")
  implicit val executionContext = system.dispatcher
  implicit val serialization = jackson.Serialization
  implicit val formats = DefaultFormats

  // curl -s -H "accept: application/json" http://httpbin.org/stream/99 | jq
  val uri = s"http://httpbin.org/stream/99"
  val futureResponse = Http().singleRequest(HttpRequest(uri = uri))

  val strictEntity:Future[HttpEntity.Strict] = futureResponse.flatMap{_.entity.toStrict(3.seconds)}

  val futureResult: Future[List[JValue]] = futureResponse.flatMap { response =>
    response.entity.dataBytes
      .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 256))
      .map(buf => parse(buf.decodeString("UTF-8")))
      .runFold(List.empty[JValue]) {case (acc,jv) => jv::acc}
  }

  val out = for {
    results <- futureResult
  } yield {
    results.foreach{result => println(pretty(render(result))) }
  }

  // Do not exit before the future has completed ;)
  def andWait():Unit = Await.ready(out, 10.seconds)

}

TestThat.andWait()