-
-
Save claudio-scandura/a8b0011beddb6c8d41f81a551003dc9f to your computer and use it in GitHub Desktop.
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?
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/
Do you think that maybe sending event to actor could be performed outside
complete
directive?