SSE Streaming Actor

StreamActor.scala

package com.besterdesigns.actors

import java.util.Calendar
import _root_.akka.actor.{Actor, Props}
import _root_.akka.stream.scaladsl.SourceQueueWithComplete
import com.besterdesigns.models.Json4sFormat
import scala.concurrent.duration._

case object StartClock

object StreamingEventSourceActor {

  def name = "StreamingEventSourceActor"

  def props(source: SourceQueueWithComplete[String]): Props = Props(new StreamingEventSourceActor(source))

  case class UpdateDashboard(data: String)
}

class StreamingEventSourceActor(source: SourceQueueWithComplete[String])
  extends Actor
  with Json4sFormat {
  implicit val ec = context.system.dispatcher
  
  override def receive: Receive = {
    case StartClock => {
      //send periodic date/time to all browsers
      val dt = Calendar.getInstance().getTime()
      val response = s"""{"msg":"${dt}"}"""
      context.system.scheduler.scheduleOnce(1.second) {
        self ! StartClock
      }
      source.offer(response)
    }
      
    case msg: StreamingEventSourceActor.UpdateDashboard => {
      // send broadcase to all connect browsers. Triggered by PUT method in Rest end point PUT events
      val testMessage =
        """{"msg":"First Message"}
          |""".stripMargin
      source.offer(testMessage)
    }
  }
}