Initial Commit

Author: Brent Johnson

.gitignore

@@ -1,2 +1,3 @@
 *.class
 *.log
+.idea/**

README.md

@@ -0,0 +1,78 @@
+# SP-Cron
+
+A small library to define functions that run on a cron schedule
+using [fs2-cron](https://github.com/fthomas/fs2-cron). Using 
+[redis](redis.io) to create a lock per job preventing the job
+from running on multiple instances at the same time.
+
+## Usage
+
+Running cron jobs is as simple as creating a Runner and passing
+it JobDefinitions. The following example defines a single job
+which prints the time every minute. The example assumes that
+`redis` defines a connection to [redis](redis.io). See the
+LockingWorker section for implementing a LockingWorker.    
+
+```                                 
+val lock = new RedisLockingWorker(redis)
+val runner = new Runner(lock)
+val jobs = Seq(
+    JobDefinition(
+        "0 * * ? * * ",
+        "test2",
+        IO({ println(LocalDateTime.now()) }),
+        45
+    )
+)
+runner.start(jobs)
+```                                                     
+
+Likely, a job will have side effects involving futures. Because
+the JobDescription expects an `IO[Unit]` the future will either
+need to be resolved, through an Await for example, or by being
+converted to an IO. A future can be converted to an IO with
+Catbird with, `rerunnableToIO(Rerunnable.fromFuture(F))`, where
+F is the Twitter Future. 
+
+### LockingWorker
+
+The LockingWorker trait must be implemented and provided to the
+Runner. LockingWorker expects implementations of 4 redis functions:
+
+* setNx: to set a key only if it doesn't exist. A UUID is 
+generated for each instantiation of the runner which is set
+as the key value.
+* expire: to expire the key based on the job's ttl
+* get: to get the value of the key
+* eval: which is used by threads that do not secure the log
+to verify a ttl is set on the key. It does this by calling
+ttl on the key and calling expire if no ttl is set.
+
+An example implementation using 
+[Finatra](https://twitter.github.io/finatra/) and 
+[Catbird](https://github.com/travisbrown/catbird) is provided 
+below.
+
+```
+class RedisLockingWorker(redis: RedisClient) extends LockingWorker {
+    val keyPrefix = "cron:namespace:"
+
+    override def setNx(key: String, value: String): IO[Boolean] =
+      redis.setNx(keyPrefix + key, value).map(b => Boolean.unbox(b))
+
+    override def expire(key: String, ttl: Long): IO[Boolean] =
+      redis.expire(keyPrefix + key, ttl).map(b => Boolean.unbox(b))
+
+    override def get(key: String): IO[String] =
+      redis.get(keyPrefix + key).map(buf => optBufToStr(buf))
+
+    override def eval(script: String, key: String, ttl: Long): IO[Boolean] =
+      redis.eval(script, Seq(keyPrefix + key), Seq(ttl.toString)).map(_ => true)
+
+    def optBufToStr(b: Option[Buf]): String = Utf8.unapply(b.get).get
+    implicit val contextShift: ContextShift[IO] = IO.contextShift(ExecutionContext.global)
+    implicit def futureToIO[A](f: Future[A]): IO[A] = rerunnableToIO(Rerunnable.fromFuture(f))
+    implicit def strToBuf(s: String): Buf = Utf8.apply(s)
+    implicit def seqStrToSeqBuf(strings: Seq[String]): Seq[Buf] = strings.map(strToBuf)
+}
+```

build.sbt

@@ -0,0 +1,14 @@
+name := "sp-cron"
+organization := "com.signalpath"
+version := "1.0"
+scalaVersion := "2.12.11"
+
+libraryDependencies ++= Seq(
+  "co.fs2" %% "fs2-core" % "2.2.1",
+  "com.github.alonsodomin.cron4s" %% "cron4s-core" % "0.6.0",
+  "eu.timepit" %% "fs2-cron-core" % "0.2.2",
+  "org.typelevel" %% "cats-core" % "2.0.0",
+  "org.scalatest" %% "scalatest" % "3.1.1" % Test,
+  "org.scalamock" %% "scalamock" % "4.4.0" % Test,
+  "org.scalatest" %% "scalatest" % "3.1.0" % Test
+)

src/main/scala/com/signalpath/spcron/LockingWorker.scala

@@ -0,0 +1,12 @@
+package com.signalpath.spcron
+
+import cats.effect.IO
+
+trait LockingWorker {
+
+  def setNx(key: String, value: String): IO[Boolean]
+  def expire(key: String, ttl: Long): IO[Boolean]
+  def get(key: String): IO[String]
+  def eval(script: String, key: String, ttl: Long): IO[Boolean]
+
+}

src/main/scala/com/signalpath/spcron/Runner.scala

@@ -0,0 +1,64 @@
+package com.signalpath.spcron
+
+import cats.effect.{Async, IO, Timer}
+import cron4s.Cron
+import eu.timepit.fs2cron.awakeEveryCron
+import fs2.{Stream => fs2Stream}
+
+import scala.concurrent.ExecutionContext
+
+case class JobDefinition(
+  cronExpression: String,
+  name: String,
+  job: IO[Unit],
+  ttl: Long) {
+
+  def cron: cron4s.CronExpr = Cron.unsafeParse(cronExpression)
+}
+
+class Runner(lock: LockingWorker) {
+
+  val runnerId = java.util.UUID.randomUUID().toString
+
+  def start(jobs: Seq[JobDefinition]): Unit = {
+    implicit val timer: Timer[IO] = IO.timer(ExecutionContext.global)
+    val streams = jobs.map { job =>
+      awakeEveryCron[IO](job.cron) >> worker(job.job, job.name, job.ttl)
+    }
+    streams
+      //.dropRight(1)
+      .map(j => j.compile.drain.unsafeRunAsync((cb: Either[Throwable, Unit]) => Unit))
+    ()
+  }
+
+  def worker(task: IO[Unit], name: String, ttl: Long): fs2.Stream[IO, fs2.INothing] = {
+    val async: IO[Unit] = Async[IO].async { cb =>
+      getLock(name, ttl).flatMap { isOwner =>
+        if(isOwner) task
+        else IO(())
+      }.unsafeRunSync()
+    }
+    fs2Stream.eval_(IO(async.unsafeRunAsync{
+      case Left(error) =>
+        throw error
+        ()
+      case Right(unit) => unit
+    }))
+  }
+
+  def getLock(name: String, ttl: Long): IO[Boolean] = {
+    lock.setNx(name, runnerId).flatMap { setnxResult =>
+      if (setnxResult) {
+        lock.expire(name, ttl).map(_ => true)
+      } else {
+        lock.get(name).flatMap { owner =>
+          val ownerCheck = runnerId.equalsIgnoreCase(owner)
+          lock.eval(
+            "local ttl = redis.call('ttl', KEYS[1]) if ttl <= 0 then redis.call('expire', KEYS[1], ARGV[1]) end return ttl",
+            name,
+            ttl).map(_ => ownerCheck)
+        }
+      }
+    }
+  }
+}

src/test/scala/com/signalpath/spcron/RunnerTest.scala

@@ -0,0 +1,90 @@
+package com.signalpath.spcron
+
+import cats.effect.{IO, Timer}
+import eu.timepit.fs2cron.awakeEveryCron
+import org.scalamock.scalatest.MockFactory
+import org.scalatest.funspec.AnyFunSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.Duration
+
+class RunnerTest extends AnyFunSpec with MockFactory with Matchers {
+
+  describe("getLock") {
+    describe("when the key does not exist") {
+      it("should set a ttl on key and return true") {
+        val mockLock = mock[LockingWorker]
+        (mockLock.setNx _).expects("test", *).returning(IO(true)).once()
+        (mockLock.expire _).expects("test", 42).returning(IO(true)).once()
+        val cron = new Runner(mockLock)
+        val actual = cron.getLock("test", 42).unsafeRunSync()
+        actual should equal(true)
+      }
+    }
+
+    describe("when the key already exists") {
+      it("should return true when key is owned by runner") {
+        val mockLock = mock[LockingWorker]
+        val cron = new Runner(mockLock)
+        (mockLock.setNx _).expects("test", *).returning(IO(false)).once()
+        (mockLock.get _).expects("test").returning(IO(cron.runnerId)).once()
+        (mockLock.eval _).expects(*, "test", 42).returning(IO(true)).once()
+        val actual = cron.getLock("test", 42).unsafeRunSync()
+        actual should equal(true)
+      }
+
+      it("should return false when key is owned by a different runner") {
+        val mockLock = mock[LockingWorker]
+        val cron = new Runner(mockLock)
+        (mockLock.setNx _).expects("test", *).returning(IO(false)).once()
+        (mockLock.get _).expects("test").returning(IO("nope")).once()
+        (mockLock.eval _).expects(*, "test", 42).returning(IO(true)).once()
+        val actual = cron.getLock("test", 42).unsafeRunSync()
+        actual should equal(false)
+      }
+    }
+  }
+
+  describe("worker") {
+    implicit val timer: Timer[IO] = IO.timer(ExecutionContext.global)
+    describe("when it secures the lock") {
+      it("should run the job") {
+        var wasRun = false
+        val j = IO({
+          wasRun = true
+          ()
+        })
+        val job = JobDefinition("* * * ? * *", "test", j, 1)
+        val mockLock = mock[LockingWorker]
+        val cron = new Runner(mockLock)
+        (mockLock.setNx _).expects(job.name, *).returning(IO(true)).atLeastOnce()
+        (mockLock.expire _).expects(job.name, job.ttl).returning(IO(true)).atLeastOnce()
+        val runner = new Runner(mockLock)
+        val worker = runner.worker(job.job, job.name, job.ttl)
+        (awakeEveryCron[IO](job.cron) >> worker).compile.drain.unsafeRunTimed(Duration(1, "second"))
+        wasRun should equal(true)
+      }
+    }
+    describe("when it does not secure the lock") {
+      it("should not run the job") {
+        var wasRun = false
+        val j = IO({
+          wasRun = true
+          ()
+        })
+        val job = JobDefinition("* * * ? * *", "test", j, 1)
+        val mockLock = mock[LockingWorker]
+        val cron = new Runner(mockLock)
+        (mockLock.setNx _).expects(job.name, *).returning(IO(false)).atLeastOnce()
+        (mockLock.get _).expects(job.name).returning(IO("nope")).atLeastOnce()
+        (mockLock.eval _).expects(*, job.name, job.ttl).returning(IO(true)).atLeastOnce()
+        val runner = new Runner(mockLock)
+        val worker = runner.worker(job.job, job.name, job.ttl)
+        (awakeEveryCron[IO](job.cron) >> worker).compile.drain.unsafeRunTimed(Duration(1, "second"))
+        wasRun should equal(false)
+      }
+    }
+  }
+
+}