package wow.doge.mygame.game.entities import scala.concurrent.duration._ import scala.util.Failure import scala.util.Success import akka.actor.typed.ActorRef import akka.actor.typed.Behavior import akka.actor.typed.LogOptions import akka.actor.typed.PostStop import akka.actor.typed.SupervisorStrategy import akka.actor.typed.scaladsl.ActorContext import akka.actor.typed.scaladsl.Behaviors import akka.util.Timeout import com.typesafe.scalalogging.Logger import monix.reactive.Observable import monix.reactive.OverflowStrategy import monix.reactive.subjects.ConcurrentSubject import org.slf4j.event.Level import wow.doge.mygame.Dispatchers import wow.doge.mygame.executors.Schedulers.AsyncScheduler import wow.doge.mygame.game.entities.StatsActor import wow.doge.mygame.implicits._ import wow.doge.mygame.subsystems.events.EventBus import wow.doge.mygame.subsystems.events.EventsModule.GameEventBus import wow.doge.mygame.subsystems.events.PlayerEvent import wow.doge.mygame.subsystems.events.TickEvent import wow.doge.mygame.subsystems.events.TickEvent.RenderTick import wow.doge.mygame.subsystems.movement.ImMovementActor import monix.eval.Coeval import monix.execution.AsyncQueue object PlayerActorSupervisor { type Ref = ActorRef[PlayerActorSupervisor.Command] sealed trait Status object Status { case object Alive extends Status case object Dead extends Status } sealed trait Command case class TakeDamage(value: CharacterStats.DamageHealth) extends Command case class TakeDamage2( value: CharacterStats.DamageHealth, replyTo: ActorRef[Unit] ) extends Command case class Heal(value: CharacterStats.HealHealth) extends Command case class CurrentStats(replyTo: ActorRef[CharacterStats]) extends Command case class GetStatus(replyTo: ActorRef[Status]) extends Command case class GetStatsObservable(replyTo: ActorRef[Observable[CharacterStats]]) extends Command case class GetStatsObservable2(replyTo: ActorRef[Observable[CharacterStats]]) extends Command private case object Die extends Command private case class DamageResponse(response: (Boolean, CharacterStats)) extends Command private case class DamageResponse2( response: (Boolean, CharacterStats), replyTo: ActorRef[Unit] ) extends Command // private case class InternalTakeDamage(old: Int, value: Int) extends Command private case class LogError(ex: Throwable) extends Command class Props( val playerEventBus: GameEventBus[PlayerEvent], val tickEventBus: GameEventBus[TickEvent], val imMovementActorBehavior: Behavior[ImMovementActor.Command], val scheduler: AsyncScheduler ) { def behavior = Behaviors.logMessages( LogOptions() .withLevel(Level.DEBUG) .withLogger( Logger[PlayerActorSupervisor].underlying ), Behaviors .setup[Command] { ctx => ctx.log.infoP("Starting PlayerActor") // spawn children actors val playerMovementActor = ctx.spawnN( Behaviors .supervise(imMovementActorBehavior) .onFailure[Exception]( SupervisorStrategy.restart.withLimit(2, 100.millis) ), Dispatchers.jmeDispatcher ) val playerStatsActor = ctx.spawnN( new StatsActor.Props( CharacterStats.Health(100), CharacterStats.Stamina(100) ).behavior ) val playerMovementEl = ctx.spawnN( Behaviors .supervise(PlayerMovementEventListener(playerMovementActor)) .onFailure[Exception]( SupervisorStrategy.restart.withLimit(2, 100.millis) ) ) val renderTickEl = { val behavior: Behavior[RenderTick.type] = Behaviors.setup(ctx => Behaviors .receiveMessage[RenderTick.type] { case RenderTick => playerMovementActor ! ImMovementActor.Tick // playerCameraActor ! PlayerCameraActor.Tick Behaviors.same } .receiveSignal { case (_, PostStop) => ctx.log.infoP("stopped") Behaviors.same } ) ctx.spawn(behavior, "playerMovementTickListener") } //init listeners playerEventBus ! EventBus.Subscribe(playerMovementEl) tickEventBus ! EventBus.Subscribe(renderTickEl) new PlayerActorSupervisor( ctx, this, Children(playerMovementActor, playerStatsActor), ConcurrentSubject.publish( OverflowStrategy.DropOldAndSignal( 50, dropped => Coeval.pure(None) ) )(scheduler.value), AsyncQueue.bounded(10)(scheduler.value) ).aliveState } ) } case class Children( movementActor: ActorRef[ImMovementActor.Command], statsActor: ActorRef[StatsActor.Command] ) } class PlayerActorSupervisor( ctx: ActorContext[PlayerActorSupervisor.Command], props: PlayerActorSupervisor.Props, children: PlayerActorSupervisor.Children, statsSubject: ConcurrentSubject[CharacterStats, CharacterStats], statsQueue: AsyncQueue[CharacterStats] ) { import PlayerActorSupervisor._ implicit val timeout = Timeout(1.second) val aliveState = Behaviors .receiveMessage[Command] { case TakeDamage(value) => // children.movementActor ! ImMovementActor.MovedDown(true) // ctx.ask(children.statsActor, StatsActor.CurrentStats(_)) { // case Success(status) => InternalTakeDamage(status.hp, value) // case Failure(ex) => LogError(ex) // } ctx.ask(children.statsActor, StatsActor.TakeDamageResult(value, _)) { case Success(response) => DamageResponse(response) case Failure(ex) => LogError(ex) } Behaviors.same case TakeDamage2(value, replyTo) => // children.movementActor ! ImMovementActor.MovedDown(true) // ctx.ask(children.statsActor, StatsActor.CurrentStats(_)) { // case Success(status) => InternalTakeDamage(status.hp, value) // case Failure(ex) => LogError(ex) // } ctx.ask(children.statsActor, StatsActor.TakeDamageResult(value, _)) { case Success(response) => DamageResponse2(response, replyTo) case Failure(ex) => LogError(ex) } Behaviors.same case CurrentStats(replyTo) => // ctx.ask(children.statsActor, StatsActor.CurrentStats()) children.statsActor ! StatsActor.CurrentStats(replyTo) Behaviors.same case Heal(value) => children.statsActor ! StatsActor.HealResult(value) Behaviors.same case GetStatus(replyTo) => replyTo ! Status.Alive Behaviors.same // case _ => Behaviors.unhandled // case InternalTakeDamage(hp, damage) => // if (hp - damage <= 0) dead // else { // children.statsActor ! StatsActor.TakeDamage(damage) // Behaviors.same // } case GetStatsObservable(replyTo) => replyTo ! statsSubject Behaviors.same case GetStatsObservable2(replyTo) => import monix.{eval => me} replyTo ! Observable.repeatEvalF( me.Task.deferFuture(statsQueue.poll()) ) Behaviors.same case DamageResponse(response) => response match { case (dead, state) => if (dead) ctx.self ! Die statsSubject.onNext(state) } Behaviors.same case DamageResponse2(response, replyTo) => response match { case (dead, stats) => if (dead) ctx.self ! Die statsQueue .offer(stats) .foreach(_ => replyTo ! ())(props.scheduler.value) } Behaviors.same case Die => deadState case LogError(ex) => ctx.log.error(ex.getMessage) Behaviors.same } .receiveSignal { case (_, PostStop) => ctx.log.infoP("stopped") statsSubject.onComplete() Behaviors.same } val deadState = Behaviors .receiveMessage[Command] { // case TakeDamage(value) => // children.statsActor ! StatsActor.TakeDamage(value) // // children.movementActor ! ImMovementActor.MovedDown(true) // Behaviors.same // case CurrentStats(replyTo) => // // ctx.ask(children.statsActor, StatsActor.CurrentStats()) // children.statsActor ! StatsActor.CurrentStats(replyTo) // Behaviors.same // case Heal(_) => // Behaviors.same case CurrentStats(replyTo) => // ctx.ask(children.statsActor, StatsActor.CurrentStats()) children.statsActor ! StatsActor.CurrentStats(replyTo) Behaviors.same case GetStatus(replyTo) => replyTo ! Status.Dead Behaviors.same case _ => Behaviors.unhandled } .receiveSignal { case (_, PostStop) => ctx.log.infoP("stopped") statsSubject.onComplete() Behaviors.same } }