akka-http отправляет непрерывный частичный HTTP-ответ (поток)

Question

У меня есть этот грубый тестовый пример с akka-http клиент и сервер.

Server.scala:

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import scala.concurrent.Future

class Server extends Runnable {

    def run() = {

        implicit val system = ActorSystem("server")
        implicit val materializer = ActorMaterializer()

        val serverSource = Http().bind(interface = "localhost", port = 8200)

        val requestHandler: HttpRequest => HttpResponse = {
            case HttpRequest(GET, Uri.Path("/stream"), _, _, _) =>
                HttpResponse(entity = HttpEntity(MediaTypes.`text/plain`, "test"))
        }

        val bindingFuture: Future[Http.ServerBinding] = serverSource.to(Sink.foreach { connection =>
            connection handleWithSyncHandler requestHandler
        }).run()

    }

}

Client.scala:

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{Uri, HttpRequest}
import akka.stream.ActorMaterializer

object Client extends App {

    implicit val system = ActorSystem("client")
    import system.dispatcher

    new Thread(new Server).start()

    implicit val materializer = ActorMaterializer()
    val source = Uri("http://localhost:8200/stream")
    val finished = Http().singleRequest(HttpRequest(uri = source)).flatMap { response =>
        response.entity.dataBytes.runForeach { chunk =>
            println(chunk.utf8String)
        }
    }

}

В данный момент Server просто отвечает одним "тестом".

Как я могу изменить HttpResponse в Server отправлять "тест" как chunked (поток) в бесконечный цикл каждую 1 секунду?

Answer 1

Нашел ответ.

Server.scala:

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Source, Sink}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import scala.concurrent.Future
import scala.concurrent.duration._

class Server extends Runnable {

    def run() = {

        implicit val system = ActorSystem("server")
        implicit val materializer = ActorMaterializer()

        val serverSource = Http().bind(interface = "localhost", port = 8200)

        val requestHandler: HttpRequest => HttpResponse = {
            case HttpRequest(GET, Uri.Path("/stream"), _, _, _) =>
                HttpResponse(entity = HttpEntity.Chunked(ContentTypes.`text/plain`, Source(0 seconds, 1 seconds, "test")))
        }

        val bindingFuture: Future[Http.ServerBinding] = serverSource.to(Sink.foreach { connection =>
            connection handleWithSyncHandler requestHandler
        }).run()

    }

}