AkkHttpChunked.scala

AkkaHttpChunked.scala

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshalling.Marshaller
import akka.http.scaladsl.marshalling.ToResponseMarshallable.apply
import akka.http.scaladsl.marshalling.ToResponseMarshaller
import akka.http.scaladsl.model.{ ContentTypes, HttpEntity }
import akka.http.scaladsl.model.HttpEntity.ChunkStreamPart
import akka.http.scaladsl.model.HttpResponse
import akka.http.scaladsl.server.Directive.addByNameNullaryApply
import akka.http.scaladsl.server.Directives.{ _segmentStringToPathMatcher, complete, get, path }
import akka.http.scaladsl.server.RouteResult.route2HandlerFlow
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source

object StreamNumbers extends App {
  implicit val system = ActorSystem("my-system")
  implicit val materializer = ActorMaterializer()

  implicit val toResponseMarshaller: ToResponseMarshaller[Source[Int, Any]] =
    Marshaller.opaque { items =>
      val data = items.map(item => ChunkStreamPart(item.toString + "\n"))
      HttpResponse(entity = HttpEntity.Chunked(ContentTypes.`application/json`, data))
    }

  def newDataStream(): Stream[Int] = Stream.from(1)

  val route =
    path("numbers") {
      get {
        complete {
          Source(newDataStream())
        }
      }
    }

  val bindingFuture = Http().bindAndHandle(route, "localhost", 8080)

  println(s"Server online at http://localhost:8080/\nPress RETURN to stop...")
  scala.io.StdIn.readLine()

  import system.dispatcher // for the future transformations
  bindingFuture
    .flatMap(_.unbind()) // trigger unbinding from the port
    .onComplete(_ ⇒ system.terminate()) // and shutdown when done
}