Akka-http example of SSE using an Actor as the source of the events Stream.

SseApp.scala

import akka.actor.{Actor, ActorSystem, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.model.sse.ServerSentEvent
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.Route
import akka.stream._
import akka.stream.scaladsl.{BroadcastHub, Keep, Source, SourceQueueWithComplete}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

object SseApp extends App {

  import akka.http.scaladsl.marshalling.sse.EventStreamMarshalling._

  implicit val actorSystem = ActorSystem()
  implicit val mat = ActorMaterializer()

  val (sourceQueue, eventsSource) = Source.queue[String](Int.MaxValue, OverflowStrategy.backpressure)
    .delay(1.seconds, DelayOverflowStrategy.backpressure)
    .map(message => ServerSentEvent(message))
    .keepAlive(1.second, () => ServerSentEvent.heartbeat)
    .toMat(BroadcastHub.sink[ServerSentEvent])(Keep.both)
    .run()

  val streamingActor = actorSystem.actorOf(Props(classOf[StreamingActor], sourceQueue))

  def route: Route = {

    path("events") {
      get {
        complete {
          eventsSource
        }
      } ~ put {
        entity(as[String]) { event =>
          complete {
            streamingActor ! event
            StatusCodes.OK
          }
        }
      }
    }
  }

  Http().bindAndHandle(route, "0.0.0.0", 9999)

  class StreamingActor(source: SourceQueueWithComplete[String]) extends Actor {
    override def receive: Receive = {
      case msg: String => source.offer(msg)
    }
  }

}

What happens and how does one deal with EventSource closing on browser side - ie eventSource.close() method is called for whatever reason?

I have the same question. Did you figure out how to stop the actor when eventSource.close()? @colinbes

@kelly-xuxixi, I haven't yet and do need to get back to it - hopefully next week. If I learn anything I will update post.

I was playing with this today and I had difficulty making the eventsSource notice a browser termination, until I replaced

    .toMat(BroadcastHub.sink[ServerSentEvent])(Keep.both)    
    .run()

with

    .preMaterialize()

Then my various .watchCompletion and .watchTermination blocks started working.

This article helped, you have to kind of squint to translate the java to scala and the Source.actorRef to the Source.queue. https://www.linkedin.com/pulse/managing-eventsource-connections-akka-actors-play-28-akhilesh-gupta/