jmonkey-test/src/main/scala/wow/doge/mygame/subsystems/events/EventBus.scala

package wow.doge.mygame.subsystems.events

import scala.reflect.ClassTag

import akka.actor.typed.ActorRef
import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl.Behaviors
import akka.event.EventStream
import monix.reactive.Observable
import monix.reactive.OverflowStrategy
import monix.execution.cancelables.SingleAssignCancelable
import monix.execution.Ack
import akka.util.Timeout
import akka.actor.typed.Scheduler
import akka.actor.typed.SpawnProtocol
import scala.util.Random
import akka.actor.typed.scaladsl.AskPattern._
import monix.execution.Cancelable
import wow.doge.mygame.utils.AkkaUtils
import wow.doge.mygame.implicits._
import monix.bio.UIO
import wow.doge.mygame.subsystems.events.EventBus.ObservableSubscription

/**
  * A (typed) event bus
  * Copied (and repurposed) from Akka's EventStream
  */
object EventBus {
  sealed trait Command[A]
  final case class Publish[A, E <: A](
      event: E,
      publisherName: String
  ) extends Command[A]

  final case class Subscribe[A, E <: A](subscriber: ActorRef[E])(implicit
      classTag: ClassTag[E]
  ) extends Command[A] {
    def topic: Class[_] = classTag.runtimeClass
  }

  final case class Unsubscribe[A, E <: A](subscriber: ActorRef[E])
      extends Command[A]

  final case class ObservableSubscription[A, E <: A](
      replyTo: ActorRef[Observable[E]]
  )(implicit
      classTag: ClassTag[E]
  ) extends Command[A] {
    def ct = classTag
  }

  def apply[A: ClassTag]()(implicit
      timeout: Timeout,
      spawnProtocol: ActorRef[SpawnProtocol.Command]
  ): Behavior[EventBus.Command[A]] =
    Behaviors.setup { ctx =>
      val eventStream = new EventStream(ctx.system.classicSystem)
      implicit val scheduler = ctx.system.scheduler
      new EventBus().eventStreamBehavior(eventStream)
    }

  def observable[E](eventBus: ActorRef[EventBus.Command[E]])(implicit
      timeout: Timeout,
      scheduler: Scheduler,
      spawnProtocol: ActorRef[SpawnProtocol.Command],
      ct: ClassTag[E]
  ) =
    Observable.create[E](OverflowStrategy.DropOld(50)) { sub =>
      implicit val s = sub.scheduler
      val c = SingleAssignCancelable()
      val behavior = Behaviors.receive[E] { (ctx, msg) =>
        if (sub.onNext(msg) == Ack.Stop) {
          c.cancel()
          Behaviors.stopped
        } else Behaviors.same

      }
      val actor =
        AkkaUtils
          .spawnActorL(
            behavior,
            s"eventBusObservable-${ct.toString}-${Random.nextLong()}"
          )
          .tapError {
            case ex => UIO(sub.onError(ex))
          }

      (for {
        a <- actor
        _ <- eventBus !! Subscribe(a)
        _ <- UIO(c := Cancelable(() => eventBus ! Unsubscribe(a)))
      } yield ()).runToFuture
      c
    }

  def observable2[A, B <: A](eventBus: ActorRef[EventBus.Command[A]])(implicit
      timeout: Timeout,
      scheduler: Scheduler,
      spawnProtocol: ActorRef[SpawnProtocol.Command],
      ct: ClassTag[A],
      ct2: ClassTag[B]
  ) =
    Observable.create[B](OverflowStrategy.DropOld(50)) { sub =>
      implicit val s = sub.scheduler
      val c = SingleAssignCancelable()
      val behavior = Behaviors.receive[B] { (ctx, msg) =>
        if (sub.onNext(msg) == Ack.Stop) {
          c.cancel()
          Behaviors.stopped
        } else Behaviors.same

      }
      val actor =
        AkkaUtils
          .spawnActorL(
            behavior,
            s"eventBusObservable-${ct.toString}-${math.abs(Random.nextLong())}"
          )
          .tapError {
            case ex => UIO(sub.onError(ex))
          }

      (for {
        a <- actor
        _ <- eventBus !! Subscribe(a)
        _ <- UIO(c := Cancelable(() => eventBus ! Unsubscribe(a)))
      } yield ()).runToFuture
      c
    }

}

class EventBus[A] {
  import akka.actor.typed.scaladsl.adapter._

  private def eventStreamBehavior(
      eventStream: EventStream
  )(implicit
      timeout: Timeout,
      scheduler: Scheduler,
      spawnProtocol: ActorRef[SpawnProtocol.Command],
      ct: ClassTag[A]
  ): Behavior[EventBus.Command[A]] =
    Behaviors.setup { ctx =>
      Behaviors.receiveMessage {
        case EventBus.Publish(event, name) =>
          eventStream.publish(event)
          Behaviors.same
        case s @ EventBus.Subscribe(subscriber) =>
          eventStream.subscribe(subscriber.toClassic, s.topic)
          Behaviors.same
        case EventBus.Unsubscribe(subscriber) =>
          eventStream.unsubscribe(subscriber.toClassic)
          Behaviors.same
        case s @ ObservableSubscription(replyTo) =>
          val obs = EventBus.observable2(
            ctx.self
          )(timeout, scheduler, spawnProtocol, ct, s.ct)
          replyTo ! obs
          Behaviors.same
      }
    }
}