update triggering cron tab for all entities

fupelaqu · fupelaqu · commit 834bcdf3eab7 · 2022-12-23T07:59:43.000+01:00
diff --git a/build.sbt b/build.sbt
@@ -31,7 +31,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "scheduler"
 
-ThisBuild / version := "0.1.3"
+ThisBuild / version := "0.1.4"
 
 ThisBuild / scalaVersion := "2.12.15"
 
diff --git a/common/src/main/scala/app/softnetwork/scheduler/model/package.scala b/common/src/main/scala/app/softnetwork/scheduler/model/package.scala
@@ -8,9 +8,10 @@ import java.time.temporal.ChronoUnit
 import java.util.Date
 import com.markatta.akron.CronExpression
 import com.typesafe.scalalogging.StrictLogging
-import org.softnetwork.akka.model.Schedule
+import org.softnetwork.akka.model.{CronTab, Schedule}
 
 import scala.concurrent.duration._
+import scala.language.implicitConversions
 import scala.util.{Failure, Success, Try}
 
 /** Created by smanciot on 11/05/2021.
@@ -91,4 +92,19 @@ package object model {
       }
     }
   }
+
+  implicit def cronTabToSchedule(cronTab: CronTab): Option[Schedule] = {
+    cronTab.nextTriggered match {
+      case Some(date) =>
+        Some(
+          Schedule.defaultInstance
+            .withPersistenceId(cronTab.persistenceId)
+            .withEntityId(cronTab.entityId)
+            .withKey(cronTab.key)
+            .withScheduledDate(date)
+            .withRepeatedly(true)
+        )
+      case _ => None
+    }
+  }
 }
diff --git a/core/src/main/scala/app/softnetwork/scheduler/persistence/query/package.scala b/core/src/main/scala/app/softnetwork/scheduler/persistence/query/package.scala
@@ -26,6 +26,8 @@ package query {
 
     def schedulerDao: SchedulerDao = SchedulerDao
 
+    def forTests: Boolean = false
+
     /** Processing event
       *
       * @param event
@@ -49,7 +51,11 @@ package query {
               if (persistenceId.startsWith(schedule.persistenceId)) {
                 val entityId = persistenceId.split("\\|").last
                 if (entityId != ALL_KEY) {
-                  schedulerDao.addSchedule(schedule.withEntityId(entityId))
+                  if (forTests) {
+                    schedulerDao.addSchedule(schedule.withEntityId(entityId).withDelay(1))
+                  } else {
+                    schedulerDao.addSchedule(schedule.withEntityId(entityId))
+                  }
                 } else {
                   Future.successful(true)
                 }
@@ -69,18 +75,27 @@ package query {
         case evt: CronTabTriggeredEvent =>
           import evt._
           if (cronTab.entityId == ALL_KEY) {
-            currentPersistenceIds().runForeach(persistenceId => {
-              if (persistenceId.startsWith(cronTab.persistenceId)) {
-                val entityId = persistenceId.split("\\|").last
-                if (entityId != ALL_KEY) {
-                  schedulerDao.addCronTab(cronTab.withEntityId(entityId))
-                } else {
-                  Future.successful(true)
-                }
-              } else {
-                Future.successful(true)
-              }
-            })
+            val maybeSchedule: Option[Schedule] = cronTab
+            maybeSchedule match {
+              case Some(schedule) =>
+                currentPersistenceIds().runForeach(persistenceId => {
+                  if (persistenceId.startsWith(cronTab.persistenceId)) {
+                    val entityId = persistenceId.split("\\|").last
+                    if (entityId != ALL_KEY) {
+                      if (forTests) {
+                        schedulerDao.addSchedule(schedule.withEntityId(entityId).withDelay(1))
+                      } else {
+                        schedulerDao.addSchedule(schedule.withEntityId(entityId))
+                      }
+                    } else {
+                      Future.successful(true)
+                    }
+                  } else {
+                    Future.successful(true)
+                  }
+                })
+              case _ => Future.successful(Done)
+            }
           } else {
             triggerCronTab(cronTab).map {
               case true => Done
diff --git a/core/src/main/scala/app/softnetwork/scheduler/persistence/typed/SchedulerBehavior.scala b/core/src/main/scala/app/softnetwork/scheduler/persistence/typed/SchedulerBehavior.scala
@@ -23,8 +23,6 @@ private[scheduler] trait SchedulerBehavior
 
   lazy val schedulerId: String = SchedulerSettings.SchedulerConfig.id.getOrElse(ALL_KEY)
 
-  override val emptyState: Option[Scheduler] = None
-
   override val snapshotInterval: Int = 100
 
   private def schedulerDao: SchedulerDao = SchedulerDao
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/handlers/SampleHandler.scala b/testkit/src/main/scala/app/softnetwork/scheduler/handlers/SampleHandler.scala
@@ -0,0 +1,9 @@
+package app.softnetwork.scheduler.handlers
+
+import app.softnetwork.persistence.message.{Command, CommandResult}
+import app.softnetwork.persistence.typed.scaladsl.EntityPattern
+import app.softnetwork.scheduler.persistence.typed.SampleTypedKey
+
+trait SampleHandler extends EntityPattern[Command, CommandResult] with SampleTypedKey {}
+
+object SampleHandler extends SampleHandler
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/message/SampleMessages.scala b/testkit/src/main/scala/app/softnetwork/scheduler/message/SampleMessages.scala
@@ -0,0 +1,15 @@
+package app.softnetwork.scheduler.message
+
+import app.softnetwork.persistence.message.{Command, CommandResult, Event}
+import app.softnetwork.scheduler.model.Sample
+
+object SampleMessages {
+
+  trait SampleCommand extends Command
+  case object AddSample extends SampleCommand
+  trait SampleResult extends CommandResult
+  case object SampleAdded extends SampleResult
+  trait SampleEvent extends Event
+  case class SampleAddedEvent(sample: Sample) extends SampleEvent
+
+}
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/model/Sample.scala b/testkit/src/main/scala/app/softnetwork/scheduler/model/Sample.scala
@@ -0,0 +1,5 @@
+package app.softnetwork.scheduler.model
+
+import app.softnetwork.persistence.model.State
+
+case class Sample(uuid: String) extends State
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/persistence/query/SampleScheduleTriggered.scala b/testkit/src/main/scala/app/softnetwork/scheduler/persistence/query/SampleScheduleTriggered.scala
@@ -0,0 +1,5 @@
+package app.softnetwork.scheduler.persistence.query
+
+import org.softnetwork.akka.model.Schedule
+
+case class SampleScheduleTriggered(schedule: Schedule)
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/persistence/query/SchedulerToSampleProcessorStream.scala b/testkit/src/main/scala/app/softnetwork/scheduler/persistence/query/SchedulerToSampleProcessorStream.scala
@@ -0,0 +1,27 @@
+package app.softnetwork.scheduler.persistence.query
+
+import akka.actor.typed.eventstream.EventStream.Publish
+import app.softnetwork.persistence.message.{Command, CommandResult}
+import app.softnetwork.persistence.query.JournalProvider
+import app.softnetwork.scheduler.handlers.SampleHandler
+import org.softnetwork.akka.model.Schedule
+
+import scala.concurrent.Future
+
+trait SchedulerToSampleProcessorStream
+    extends Scheduler2EntityProcessorStream[Command, CommandResult]
+    with SampleHandler {
+  _: JournalProvider =>
+
+  override def forTests: Boolean = true
+
+  /** @param schedule
+    *   - the schedule to trigger
+    * @return
+    *   true if the schedule has been successfully triggered, false otherwise
+    */
+  override protected def triggerSchedule(schedule: Schedule): Future[Boolean] = {
+    system.eventStream.tell(Publish(SampleScheduleTriggered(schedule)))
+    Future.successful(true)
+  }
+}
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/persistence/typed/SampleBehavior.scala b/testkit/src/main/scala/app/softnetwork/scheduler/persistence/typed/SampleBehavior.scala
@@ -0,0 +1,39 @@
+package app.softnetwork.scheduler.persistence.typed
+
+import akka.actor.typed.ActorRef
+import akka.actor.typed.scaladsl.{ActorContext, TimerScheduler}
+import akka.persistence.typed.scaladsl.Effect
+import app.softnetwork.persistence.message.{Command, CommandResult, Event}
+import app.softnetwork.persistence.typed._
+import app.softnetwork.scheduler.message.SampleMessages.{AddSample, SampleAdded, SampleAddedEvent}
+import app.softnetwork.scheduler.model.Sample
+
+trait SampleBehavior extends EntityBehavior[Command, Sample, Event, CommandResult] {
+  override def persistenceId: String = "Sample"
+
+  override def handleCommand(
+    entityId: String,
+    state: Option[Sample],
+    command: Command,
+    replyTo: Option[ActorRef[CommandResult]],
+    timers: TimerScheduler[Command]
+  )(implicit context: ActorContext[Command]): Effect[Event, Option[Sample]] = {
+    command match {
+      case AddSample =>
+        Effect.persist(SampleAddedEvent(Sample(entityId))).thenRun(_ => SampleAdded ~> replyTo)
+      case _ => super.handleCommand(entityId, state, command, replyTo, timers)
+    }
+  }
+
+  override def handleEvent(state: Option[Sample], event: Event)(implicit
+    context: ActorContext[_]
+  ): Option[Sample] = {
+    event match {
+      case evt: SampleAddedEvent =>
+        Some(evt.sample)
+      case _ => super.handleEvent(state, event)
+    }
+  }
+}
+
+object SampleBehavior extends SampleBehavior
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/persistence/typed/SampleTypedKey.scala b/testkit/src/main/scala/app/softnetwork/scheduler/persistence/typed/SampleTypedKey.scala
@@ -0,0 +1,12 @@
+package app.softnetwork.scheduler.persistence.typed
+
+import akka.cluster.sharding.typed.scaladsl.EntityTypeKey
+import app.softnetwork.persistence.message.Command
+import app.softnetwork.persistence.typed.CommandTypeKey
+
+import scala.reflect.ClassTag
+
+trait SampleTypedKey extends CommandTypeKey[Command] {
+  override def TypeKey(implicit tTag: ClassTag[Command]): EntityTypeKey[Command] =
+    SampleBehavior.TypeKey
+}
diff --git a/testkit/src/main/scala/app/softnetwork/scheduler/scalatest/SchedulerWithSampleTestKit.scala b/testkit/src/main/scala/app/softnetwork/scheduler/scalatest/SchedulerWithSampleTestKit.scala
@@ -0,0 +1,32 @@
+package app.softnetwork.scheduler.scalatest
+
+import akka.actor.testkit.typed.scaladsl.TestProbe
+import akka.actor.typed.ActorSystem
+import app.softnetwork.persistence.launch
+import app.softnetwork.persistence.launch.PersistenceGuardian._
+import app.softnetwork.persistence.query.InMemoryJournalProvider
+import app.softnetwork.scheduler.persistence.query.{
+  SampleScheduleTriggered,
+  Scheduler2EntityProcessorStream,
+  SchedulerToSampleProcessorStream
+}
+import app.softnetwork.scheduler.persistence.typed.{SampleBehavior, SchedulerBehavior}
+import org.scalatest.Suite
+
+trait SchedulerWithSampleTestKit extends SchedulerTestKit { _: Suite =>
+  override def schedulerEntities: ActorSystem[_] => Seq[launch.PersistentEntity[_, _, _, _]] =
+    _ => Seq(SchedulerBehavior, SampleBehavior)
+
+  override def scheduler2EntityProcessorStreams
+    : ActorSystem[_] => Seq[Scheduler2EntityProcessorStream[_, _]] =
+    sys =>
+      Seq(new SchedulerToSampleProcessorStream with InMemoryJournalProvider {
+        override implicit def system: ActorSystem[_] = sys
+        override val tag: String = s"${SampleBehavior.persistenceId}-scheduler"
+      })
+
+  val probeSampleSchedule: TestProbe[SampleScheduleTriggered] =
+    createTestProbe[SampleScheduleTriggered]()
+  subscribeProbe(probeSampleSchedule)
+
+}
diff --git a/testkit/src/test/resources/reference.conf b/testkit/src/test/resources/reference.conf
@@ -3,7 +3,6 @@ softnetwork{
     id = "my-scheduler"
     reset-scheduler {
       initial-delay = 0
-      delay = 1
     }
     event-streams {
       entity-to-scheduler-tag = "entity-to-scheduler-"${softnetwork.scheduler.id}
diff --git a/testkit/src/test/scala/app/softnetwork/scheduler/handlers/SchedulerHandlerSpec.scala b/testkit/src/test/scala/app/softnetwork/scheduler/handlers/SchedulerHandlerSpec.scala
@@ -1,36 +1,62 @@
 package app.softnetwork.scheduler.handlers
 
+import akka.actor.testkit.typed.scaladsl.TestProbe
 import akka.actor.typed.ActorSystem
 import app.softnetwork.persistence._
-import app.softnetwork.scheduler.scalatest.SchedulerTestKit
+import app.softnetwork.scheduler.message.SampleMessages.{AddSample, SampleAdded}
+import app.softnetwork.scheduler.scalatest.SchedulerWithSampleTestKit
 import org.scalatest.wordspec.AnyWordSpecLike
 import app.softnetwork.scheduler.message._
+import app.softnetwork.scheduler.persistence.query.SampleScheduleTriggered
+import app.softnetwork.scheduler.persistence.typed.SampleBehavior
 import org.softnetwork.akka.model.{CronTab, Schedule}
 
 import scala.concurrent.ExecutionContextExecutor
 
 /** Created by smanciot on 19/03/2020.
   */
-class SchedulerHandlerSpec extends SchedulerHandler with AnyWordSpecLike with SchedulerTestKit {
+class SchedulerHandlerSpec
+    extends SchedulerHandler
+    with AnyWordSpecLike
+    with SchedulerWithSampleTestKit {
 
   implicit lazy val system: ActorSystem[Nothing] = typedSystem()
 
   implicit lazy val ec: ExecutionContextExecutor = system.executionContext
 
   "Scheduler" must {
     "add Cron Tab" in {
-      val cronTab = CronTab("p", ALL_KEY, "add", "* * * * *")
-      this !? AddCronTab(cronTab) assert {
-        case _: CronTabAdded => succeed
-        case other           => fail(other.getClass)
+      // add Sample[sample] entity
+      SampleHandler ? ("sample", AddSample) assert {
+        case SampleAdded => succeed
+        case _           => fail()
       }
+      // add cron tab for all Sample entity
+      val cronTab = CronTab(SampleBehavior.persistenceId, ALL_KEY, "cron", "* * * * *")
       this !? AddCronTab(cronTab) assert {
         case _: CronTabAdded => succeed
         case other           => fail(other.getClass)
       }
+      // trigger cron tab
+      this !? TriggerCronTab(cronTab.persistenceId, cronTab.entityId, cronTab.key) assert {
+        case _: CronTabTriggered => succeed
+        case other               => fail(other.getClass)
+      }
+      // a schedule for the Sample[sample] entity has been added to be triggered at the next cron job date
+      probeSampleSchedule.receiveMessage()
       this !? LoadScheduler assert {
         case r: SchedulerLoaded =>
-          assert(r.scheduler.cronTabs.exists(ct => ct.uuid == cronTab.uuid))
+          val scheduler = r.scheduler
+          logger.info(scheduler.toProtoString)
+          assert(scheduler.cronTabs.exists(ct => ct.uuid == cronTab.uuid))
+          scheduler.schedules.find(s =>
+            s.persistenceId == SampleBehavior.persistenceId && s.entityId == "sample" && s.key == cronTab.key
+          ) match {
+            case Some(schedule) =>
+              assert(schedule.repeatedly.getOrElse(false))
+              assert(schedule.getScheduledDate.equals(schedule.getLastTriggered))
+            case _ => fail("schedule not found")
+          }
         case _ => fail()
       }
     }
