Merge pull request #2 from SOFTNETWORK-APP/feature/akka-grpc

add support for akka grpc - update schedule flags

Author: fupelaqu

build.sbt

@@ -31,7 +31,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "scheduler"
 
-ThisBuild / version := "0.1.0"
+ThisBuild / version := "0.1.1"
 
 ThisBuild / scalaVersion := "2.12.15"
 

common/src/main/protobuf/api/schedule.proto

@@ -0,0 +1,35 @@
+syntax = "proto3";
+
+import "scalapb/scalapb.proto";
+import "model/schedule.proto";
+
+package app.softnetwork.schedule.api;
+
+option (scalapb.options) = {
+    import: "app.softnetwork.protobuf.ScalaPBTypeMappers._"
+    preserve_unknown_fields: false
+};
+
+service SchedulerServiceApi {
+    rpc AddSchedule (AddScheduleRequest) returns (AddScheduleResponse) {}
+    rpc RemoveSchedule (RemoveScheduleRequest) returns (RemoveScheduleResponse) {}
+}
+
+message AddScheduleRequest{
+    org.softnetwork.akka.model.Schedule schedule = 1;
+}
+
+message AddScheduleResponse{
+    bool succeeded = 1;
+}
+
+message RemoveScheduleRequest{
+    string persistenceId = 1;
+    string entityId = 2;
+    string key = 3;
+}
+
+message RemoveScheduleResponse{
+    bool succeeded = 1;
+}
+

common/src/main/scala/app/softnetwork/scheduler/api/GrpcClient.scala

@@ -0,0 +1,43 @@
+package app.softnetwork.scheduler.api
+
+import akka.actor.CoordinatedShutdown
+import akka.actor.typed.ActorSystem
+import akka.grpc.scaladsl.AkkaGrpcClient
+
+import app.softnetwork.persistence.typed._
+
+import scala.concurrent.ExecutionContext
+
+trait GrpcClient {
+
+  implicit def system: ActorSystem[_]
+  implicit lazy val ec: ExecutionContext = system.executionContext
+
+  def name: String
+
+  def grpcClient: AkkaGrpcClient
+}
+
+trait GrpcClientFactory[T <: GrpcClient] {
+  def name: String
+  private[this] var client: Option[T] = None
+  def init(sys: ActorSystem[_]): T
+  def apply(sys: ActorSystem[_]): T = {
+    client match {
+      case Some(value) => value
+      case _ =>
+        implicit val classicSystem: _root_.akka.actor.ActorSystem = sys
+        val shutdown = CoordinatedShutdown(classicSystem)
+        val cli = init(sys)
+        client = Some(cli)
+        shutdown.addTask(
+          CoordinatedShutdown.PhaseServiceRequestsDone,
+          s"$name-graceful-terminate"
+        ) { () =>
+          client = None
+          cli.grpcClient.close()
+        }
+        cli
+    }
+  }
+}

common/src/main/scala/app/softnetwork/scheduler/api/SchedulerClient.scala

@@ -0,0 +1,39 @@
+package app.softnetwork.scheduler.api
+
+import akka.actor.typed.ActorSystem
+import akka.grpc.GrpcClientSettings
+import app.softnetwork.schedule.api.{
+  AddScheduleRequest,
+  RemoveScheduleRequest,
+  SchedulerServiceApiClient
+}
+import org.softnetwork.akka.model.Schedule
+
+import scala.concurrent.Future
+
+trait SchedulerClient extends GrpcClient {
+
+  implicit lazy val grpcClient: SchedulerServiceApiClient =
+    SchedulerServiceApiClient(
+      GrpcClientSettings.fromConfig(name)
+    )
+
+  def addSchedule(schedule: Schedule): Future[Boolean] = {
+    grpcClient.addSchedule(AddScheduleRequest(Some(schedule))) map (_.succeeded)
+  }
+
+  def removeSchedule(persistenceId: String, entityId: String, key: String): Future[Boolean] = {
+    grpcClient.removeSchedule(RemoveScheduleRequest(persistenceId, entityId, key)) map (_.succeeded)
+  }
+
+}
+
+object SchedulerClient extends GrpcClientFactory[SchedulerClient] {
+  override val name: String = "SchedulerService"
+  override def init(sys: ActorSystem[_]): SchedulerClient = {
+    new SchedulerClient {
+      override implicit lazy val system: ActorSystem[_] = sys
+      val name: String = SchedulerClient.name
+    }
+  }
+}

common/src/main/scala/app/softnetwork/scheduler/model/package.scala

@@ -29,15 +29,18 @@ package object model {
 
   trait ScheduleDecorator {
     _: Schedule =>
-    /* flag to indicate whether we should trigger this schedule
+    /* flag to indicate whether the scheduled date has been reached or not
+     */
+    val scheduledDateReached: Boolean =
+      scheduledDate.isDefined && (now().after(getScheduledDate) || now().equals(getScheduledDate))
+    /* flag to indicate whether we could trigger this schedule
      */
     val triggerable: Boolean =
-      // the schedule has never been triggered and has no scheduled date
-      (lastTriggered.isEmpty && scheduledDate.isEmpty) ||
+      // the schedule has never been triggered or can be triggered repeatedly and has no scheduled date
+      ((lastTriggered.isEmpty || repeatedly.getOrElse(false)) && scheduledDate.isEmpty) ||
       // the schedule should be triggered at a specified date that has been reached
       // and it has not yet been triggered or has been triggered before the specified date
-      (scheduledDate.isDefined &&
-      now().after(getScheduledDate) &&
+      (scheduledDateReached &&
       (lastTriggered.isEmpty || getLastTriggered.before(getScheduledDate)))
 
     /* flag to indicate whether we should remove this schedule

core/src/main/scala/app/softnetwork/scheduler/api/SchedulerServer.scala

@@ -0,0 +1,52 @@
+package app.softnetwork.scheduler.api
+
+import akka.actor.typed.ActorSystem
+import app.softnetwork.schedule.api.{
+  AddScheduleRequest,
+  AddScheduleResponse,
+  RemoveScheduleRequest,
+  RemoveScheduleResponse,
+  SchedulerServiceApi
+}
+import app.softnetwork.scheduler.handlers.SchedulerHandler
+import app.softnetwork.scheduler.message.{
+  AddSchedule,
+  RemoveSchedule,
+  ScheduleAdded,
+  ScheduleRemoved
+}
+
+import scala.concurrent.{ExecutionContextExecutor, Future}
+
+trait SchedulerServer extends SchedulerServiceApi with SchedulerHandler {
+
+  implicit def system: ActorSystem[_]
+
+  implicit lazy val ec: ExecutionContextExecutor = system.executionContext
+
+  override def addSchedule(in: AddScheduleRequest): Future[AddScheduleResponse] = {
+    in.schedule match {
+      case Some(schedule) =>
+        !?(AddSchedule(schedule)) map {
+          case _: ScheduleAdded => AddScheduleResponse(true)
+          case _                => AddScheduleResponse()
+        }
+      case _ => Future.successful(AddScheduleResponse())
+    }
+  }
+
+  override def removeSchedule(in: RemoveScheduleRequest): Future[RemoveScheduleResponse] = {
+    !?(RemoveSchedule(in.persistenceId, in.entityId, in.key)) map {
+      case _: ScheduleRemoved => RemoveScheduleResponse(true)
+      case _                  => RemoveScheduleResponse()
+    }
+  }
+}
+
+object SchedulerServer {
+  def apply(sys: ActorSystem[_]): SchedulerServer = {
+    new SchedulerServer {
+      override implicit val system: ActorSystem[_] = sys
+    }
+  }
+}

core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerApplication.scala

@@ -1,8 +1,18 @@
 package app.softnetwork.scheduler.launch
 
+import akka.actor.typed.ActorSystem
+import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
 import app.softnetwork.api.server.launch.HealthCheckApplication
 import app.softnetwork.persistence.query.SchemaProvider
+import app.softnetwork.schedule.api.SchedulerServiceApiHandler
+
+import scala.concurrent.Future
 
 trait SchedulerApplication extends HealthCheckApplication with SchedulerGuardian {
   _: SchemaProvider =>
+
+  override def grpcServices
+    : ActorSystem[_] => Seq[PartialFunction[HttpRequest, Future[HttpResponse]]] = system =>
+    Seq(SchedulerServiceApiHandler.partial(schedulerServer(system))(system))
+
 }

core/src/main/scala/app/softnetwork/scheduler/launch/SchedulerGuardian.scala

@@ -4,6 +4,7 @@ import akka.actor.typed.ActorSystem
 import app.softnetwork.persistence.launch.{PersistenceGuardian, PersistentEntity}
 import app.softnetwork.persistence.launch.PersistenceGuardian._
 import app.softnetwork.persistence.query.{EventProcessorStream, SchemaProvider}
+import app.softnetwork.scheduler.api.SchedulerServer
 import app.softnetwork.scheduler.handlers.SchedulerDao
 import app.softnetwork.scheduler.persistence.query.{
   Entity2SchedulerProcessorStream,
@@ -48,5 +49,9 @@ trait SchedulerGuardian extends PersistenceGuardian with StrictLogging {
     }
   }
 
+  /** initialize scheduler server
+    */
+  def schedulerServer: ActorSystem[_] => SchedulerServer = sys => SchedulerServer(sys)
+
   override def initSystem: ActorSystem[_] => Unit = initSchedulerSystem
 }

testkit/build.sbt

@@ -7,5 +7,6 @@ organization := "app.softnetwork.scheduler"
 name := "scheduler-testkit"
 
 libraryDependencies ++= Seq(
-  "app.softnetwork.persistence" %% "persistence-core-testkit" % Versions.genericPersistence,
+  "app.softnetwork.api" %% "generic-server-api-testkit" % Versions.genericPersistence,
+  "app.softnetwork.persistence" %% "persistence-core-testkit" % Versions.genericPersistence
 )

testkit/src/main/scala/app/softnetwork/scheduler/api/SchedulerGrpcServer.scala

@@ -0,0 +1,13 @@
+package app.softnetwork.scheduler.api
+
+import akka.http.scaladsl.testkit.PersistenceScalatestGrpcTest
+import app.softnetwork.persistence.scalatest.InMemoryPersistenceTestKit
+import app.softnetwork.scheduler.launch.SchedulerGuardian
+import org.scalatest.Suite
+
+trait SchedulerGrpcServer
+    extends PersistenceScalatestGrpcTest
+    with SchedulerGrpcServices
+    with InMemoryPersistenceTestKit { _: Suite with SchedulerGuardian =>
+  override lazy val additionalConfig: String = grpcConfig
+}