http でじわじわとストリーミングする的な

Listener.scala

package MyFirstSprayApp

import akka.actor._
import spray.http._
import spray.http.HttpMethods._
import akka.actor.Terminated
import akka.io
import spray.can.Http
import scala.collection.immutable.HashSet
import akka.io.Tcp.Bound
import akka.io.Tcp

case class SendChunk(message: String)

object Main {
  def main(args:Array[String]): Unit = {
    val system = ActorSystem.create("MyFirstSprayApp")
    system.actorOf(Props[RootActor], name = "root")
  }
}

class RootActor extends Actor with ActorLogging{
  val listener = context.actorOf(Props[Listener], name = "listener")
  io.IO(Http)(context.system) ! Http.Bind(listener, interface = "localhost", port = 8080)

  def receive = {
    case f: Tcp.CommandFailed =>
      log.error(f.toString)
      context.system.shutdown()
  }
}

class Listener extends Actor with ActorLogging {

  var streamers: Set[ActorRef] = HashSet.empty

  def receive = {
    case Terminated(streamer) =>
      log.info("terminated streamer")
      streamers -= streamer

    case Http.Connected(remote, local) =>
      sender ! Http.Register(self)

    case HttpRequest(GET, Uri.Path("/"), _, _, _) =>
      val client = sender

      val streamer = context.actorOf(Props(new Streamer(client)))
      context watch streamer
      streamers += streamer

    case HttpRequest(POST, Uri.Path("/"), _, entity, _) =>
      log.info("posted:" + entity.asString)
      streamers.foreach(_ ! SendChunk(entity.asString))
      sender ! HttpResponse(entity = "OK")

    case req: HttpRequest =>
      log.info(req.toString)
      sender ! HttpResponse(status = StatusCode.int2StatusCode(404), entity = "Not Found")
  }
}

class Streamer(client: ActorRef) extends Actor with ActorLogging {

  context watch client
  client ! ChunkedResponseStart(HttpResponse(status = StatusCode.int2StatusCode(200)))

  def receive = {
    case Terminated(`client`) =>
      context stop self

    case _: Http.ConnectionClosed =>
      context unwatch client
      context stop self

    case SendChunk(message) =>
      client ! MessageChunk(message + "\r\n")
  }
}