Job schedule service implementation

src/main/scala/io/cronit/actors/RestTaskActor.scala

@@ -0,0 +1,12 @@
+package io.cronit.actors
+
+import akka.actor.Actor
+import io.cronit.models.Start
+
+class RestTaskActor extends Actor {
+  override def receive: Receive = {
+    case Start => {
+
+    }
+  }
+}

src/main/scala/io/cronit/models/Messages.scala

@@ -0,0 +1,9 @@
+package io.cronit.models
+
+case class Acked()
+
+case class NotAcked()
+
+case class Start()
+
+case class Finish()

src/main/scala/io/cronit/models/ScheduleOnce.scala

@@ -1,5 +1,5 @@
 package io.cronit.models
 
-import java.util.Date
+import org.joda.time.DateTime
 
-case class ScheduleOnce(runAt: Date) extends ScheduleInfo
+case class ScheduleOnce(runAt: DateTime) extends ScheduleInfo

src/main/scala/io/cronit/services/ActorSystemComponent.scala

@@ -0,0 +1,15 @@
+package io.cronit.services
+
+import akka.actor.{ActorSystem, Props}
+import io.cronit.actors.RestTaskActor
+
+
+trait ActorSystemComponent {
+  val actorSystemService: ActorSystemService
+
+  class ActorSystemService {
+    implicit lazy val actorSystem = ActorSystem("CronitActorSystem")
+    val scheduler = actorSystem.scheduler
+    val restTaskActor = actorSystem.actorOf(Props[RestTaskActor])
+  }
+}

src/main/scala/io/cronit/services/CronExpressionComponent.scala

@@ -6,6 +6,9 @@ import com.cronutils.model.time.ExecutionTime
 import com.cronutils.parser.CronParser
 import io.cronit.utils.Clock
 import io.cronit.utils.ClockUtils._
+import org.joda.time.DateTime
+
+import scala.concurrent.duration.{Duration, _}
 
 trait CronExpressionComponent {
 
@@ -18,6 +21,12 @@ trait CronExpressionComponent {
       val cron = cronParser.parse(expression)
       ExecutionTime.forCron(cron).nextExecution(Clock.asZonedDateTime).toJoda
     }
+
+    def getFiniteDurationFromNow(dateTime: DateTime) = {
+      val now = Clock.now()
+      val diff = dateTime.getMillis - now.getMillis
+      Duration(diff, MILLISECONDS)
+    }
   }
 
-}
+}

src/main/scala/io/cronit/services/JobModelBuilderComponent.scala

@@ -18,7 +18,7 @@ trait JobModelBuilderComponent {
 
       val schedulerInfo = schedulerInfoMap("type") match {
         case "CronScheduler" => CronScheduler(schedulerInfoMap("expression"))
-        case "ScheduleOnce" => ScheduleOnce(DateTime.parse(schedulerInfoMap("runAt")).toDate)
+        case "ScheduleOnce" => ScheduleOnce(DateTime.parse(schedulerInfoMap("runAt")))
       }
 
       val jobType = from("jobType")

src/main/scala/io/cronit/services/JobSchedulerComponent.scala

@@ -0,0 +1,29 @@
+package io.cronit.services
+
+import io.cronit.models.{CronScheduler, JobModel, ScheduleOnce}
+import io.cronit.utils.Clock
+import org.joda.time.DateTime
+
+import scala.concurrent.ExecutionContext.Implicits.global
+
+trait JobSchedulerComponent {
+  this: ActorSystemComponent with CronExpressionComponent =>
+  val jobSchedulerService: JobSchedulerService
+
+  class JobSchedulerService {
+    def scheduleTask(jobModel: JobModel): Unit = {
+      var executionDate: DateTime = Clock.now()
+      jobModel.scheduleInfo match {
+        case cs: CronScheduler => {
+          executionDate = cronExpressionService.getNextExecutionDate(cs.expression)
+        }
+        case so: ScheduleOnce => {
+          executionDate = so.runAt
+        }
+      }
+      val delay = cronExpressionService.getFiniteDurationFromNow(executionDate)
+      actorSystemService.scheduler.scheduleOnce(delay, actorSystemService.restTaskActor, jobModel)
+    }
+  }
+
+}

src/test/scala/io/cronit/builder/RestJobModelBuilder.scala

@@ -0,0 +1,104 @@
+package io.cronit.builder
+
+import io.cronit.models.{CronScheduler, RestJobModel, ScheduleInfo, ScheduleOnce}
+import io.cronit.utils.Clock
+
+class RestJobModelBuilder {
+
+  var method: String = ""
+  var url: String = ""
+  var body: String = ""
+  var expectedStatus: Int = 200
+  var headers: Map[String, String] = Map[String, String]()
+  var name: String = ""
+  var id: String = ""
+  var scheduleInfo: ScheduleInfo = new ScheduleInfo {}
+
+  def addHeader(key: String, value: String): RestJobModelBuilder = {
+    headers + (key -> value)
+    this
+  }
+
+  def method(method: String): RestJobModelBuilder = {
+    this.method = method
+    this
+  }
+
+  def url(url: String): RestJobModelBuilder = {
+    this.url = url
+    this
+  }
+
+  def scheduleInfo(scheduleInfo: ScheduleInfo): RestJobModelBuilder = {
+    this.scheduleInfo = scheduleInfo
+    this
+  }
+
+  def expectedStatus(expectedStatus: Int): RestJobModelBuilder = {
+    this.expectedStatus = expectedStatus
+    this
+  }
+
+  def name(name: String): RestJobModelBuilder = {
+    this.name = name
+    this
+  }
+
+  def body(body: String): RestJobModelBuilder = {
+    this.body = body
+    this
+  }
+
+  def id(id: String): RestJobModelBuilder = {
+    this.id = id
+    this
+  }
+
+
+  def build: RestJobModel = {
+    val builder = this
+    new RestJobModel {
+      override def url = builder.url
+
+      override def method = builder.method
+
+      override def body = Some(builder.body)
+
+      override def expectedStatus = builder.expectedStatus
+
+      override def headers = Some(builder.headers)
+
+      override def scheduleInfo = builder.scheduleInfo
+
+      override def name = builder.name
+
+      override def id = builder.id
+    }
+  }
+
+}
+
+object RestJobModelBuilder {
+  def aRestJobModelBuilder(): RestJobModelBuilder = new RestJobModelBuilder()
+
+  def sampleRestJobWithCronScheduler(): RestJobModel = RestJobModelBuilder.
+    aRestJobModelBuilder().
+    method("GET").
+    url("http://www.google.com").
+    addHeader("foo", "bar").
+    name("CronJob").
+    expectedStatus(200).
+    scheduleInfo(new CronScheduler("* * * * *")).
+    build
+
+  def sampleRestJobWithScheduleOnce(): RestJobModel = RestJobModelBuilder.
+    aRestJobModelBuilder().
+    method("GET").
+    url("http://www.google.com").
+    addHeader("foo", "bar").
+    name("CronJob").
+    expectedStatus(200).
+    scheduleInfo(new ScheduleOnce(Clock.now())).
+    build
+
+}

src/test/scala/io/cronit/services/CronExpressionServiceTest.scala

@@ -1,8 +1,12 @@
 package io.cronit.services
 
 import io.cronit.utils.{Clock, DateUtils}
+import org.joda.time.DateTime
 import org.scalatest.{FlatSpec, Matchers}
 
+import scala.concurrent.duration.{Duration, _}
+
+
 class CronExpressionServiceTest extends FlatSpec with Matchers {
 
   val cronExpressionComponent = new CronExpressionComponent() {
@@ -29,4 +33,15 @@ class CronExpressionServiceTest extends FlatSpec with Matchers {
 
   }
 
+  it should "get finite duration between two dates" in {
+
+    Clock.freeze(DateUtils.localDate("22032017"))
+
+    val finiteDuration = cronExpressionComponent.cronExpressionService.getFiniteDurationFromNow(DateTime.parse("2017-03-23T08:31:22+00:00"))
+    finiteDuration shouldBe Duration(124282000, MILLISECONDS)
+
+    Clock.unfreeze()
+
+  }
+
 }

src/test/scala/io/cronit/services/JobModelBuilderServiceTest.scala

@@ -91,7 +91,7 @@ class JobModelBuilderServiceTest extends FlatSpec with Matchers
     jobModel.name shouldEqual "jobName"
     jobModel.group shouldEqual "Default"
     jobModel.scheduleInfo shouldBe a[ScheduleOnce]
-    jobModel.scheduleInfo.asInstanceOf[ScheduleOnce].runAt shouldEqual DateTime.parse("2016-10-30T00:00:00.000Z").toDate
+    jobModel.scheduleInfo.asInstanceOf[ScheduleOnce].runAt shouldEqual DateTime.parse("2016-10-30T00:00:00.000Z")
 
     val restJobModel = jobModel.asInstanceOf[RestJobModel]
     restJobModel.url shouldEqual "http://cronscheduler.it"

src/test/scala/io/cronit/services/JobSchedulerServiceTest.scala

@@ -0,0 +1,51 @@
+package io.cronit.services
+
+import akka.actor.{ActorRef, Scheduler}
+import akka.testkit.TestProbe
+import io.cronit.builder.RestJobModelBuilder
+import io.cronit.utils.{Clock, DateUtils}
+import org.mockito.Mockito._
+import org.scalatest.{FlatSpecLike, Matchers}
+import org.specs2.mock.Mockito
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration._
+
+class JobSchedulerServiceTest extends ActorSystemComponent with FlatSpecLike
+  with Matchers
+  with Mockito
+  with JobSchedulerComponent
+  with CronExpressionComponent {
+
+  override val jobSchedulerService: JobSchedulerService = new JobSchedulerService
+  override val cronExpressionService: CronExpressionService = mock[CronExpressionService]
+  override val actorSystemService: ActorSystemService = new ActorSystemService {
+    override val scheduler: Scheduler = mock[Scheduler]
+    override val restTaskActor: ActorRef = TestProbe().ref
+  }
+
+  it should "calculate next execution delay when scheduling a task" in {
+    Clock.freeze(DateUtils.toDate("20180202"))
+
+    val restJob = RestJobModelBuilder.sampleRestJobWithCronScheduler
+    when(cronExpressionService.getNextExecutionDate("* * * * *")).thenReturn(Clock.now)
+    when(cronExpressionService.getFiniteDurationFromNow(Clock.now())).thenReturn(5 minutes)
+
+    jobSchedulerService.scheduleTask(restJob)
+
+    verify(actorSystemService.scheduler).scheduleOnce(5 minutes, actorSystemService.restTaskActor, restJob)
+    Clock.unfreeze()
+  }
+
+  it should "calculate next execution delay when schedule once task" in {
+    Clock.freeze(DateUtils.toDate("20180202"))
+
+    val restJob = RestJobModelBuilder.sampleRestJobWithScheduleOnce
+    when(cronExpressionService.getFiniteDurationFromNow(Clock.now())).thenReturn(15 minutes)
+
+    jobSchedulerService.scheduleTask(restJob)
+
+    verify(actorSystemService.scheduler).scheduleOnce(15 minutes, actorSystemService.restTaskActor, restJob)
+    Clock.unfreeze()
+  }
+}