Simplify Job: remove allJobStatus, jobStatus and writeStatus.

allJobStatus and jobStatus were just used by JobManager and are
now implemented just there (using jobStatusRepo).
writeStatus is extracted as trait WriteStatus and can be mixed in if
needed.

Now Job no longer depends on the JobStatusRepository which is another
benefit (less dependencies).

src/main/scala/de/kaufhof/hajobs/ActorJob.scala

@@ -3,6 +3,7 @@ package de.kaufhof.hajobs
 import akka.actor._
 
 import scala.concurrent.Future
+
 import scala.concurrent.duration._
 import scala.language.postfixOps
 
@@ -16,11 +17,10 @@ import scala.language.postfixOps
 class ActorJob(jobType: JobType,
                props: JobContext => Props,
                system: ActorSystem,
-               jobStatusRepository: JobStatusRepository,
                retriggerCount: Int = 0,
                cronExpression: Option[String] = None,
                lockTimeout: FiniteDuration = 60 seconds)
-  extends Job(jobType, jobStatusRepository, retriggerCount, cronExpression, lockTimeout) {
+  extends Job(jobType, retriggerCount, cronExpression, lockTimeout) {
 
   private var actor: Option[ActorRef] = None
 

src/main/scala/de/kaufhof/hajobs/Job.scala

@@ -29,7 +29,6 @@ case class JobContext(jobId: UUID, triggerId: UUID, finishCallback: () => Unit)
  * Base class for jobs, wraps job execution and status.
  */
 abstract class Job(val jobType: JobType,
-                   jobStatusRepository: JobStatusRepository,
                    val retriggerCount: Int,
                    val cronExpression: Option[String] = None,
                    val lockTimeout: FiniteDuration = 60 seconds) {
@@ -53,12 +52,4 @@ abstract class Job(val jobType: JobType,
    */
   def cancel()
 
-  def allJobStatus: Future[List[JobStatus]] = jobStatusRepository.list(jobType)
-
-  def jobStatus(jobId: UUID): Future[Option[JobStatus]] = jobStatusRepository.get(jobType, jobId)
-
-  protected def writeStatus(jobState: JobState, content: Option[JsValue] = None)(implicit jobContext: JobContext): Future[JobStatus] = {
-    val status = JobStatus(jobContext.triggerId, jobType, jobContext.jobId, jobState, JobStatus.stateToResult(jobState), DateTime.now(), content)
-    jobStatusRepository.save(status)
-  }
 }

src/main/scala/de/kaufhof/hajobs/JobManager.scala

@@ -21,12 +21,13 @@ import scala.util.control.NonFatal
  */
 class JobManager(managedJobs: => Jobs,
                  lockRepo: LockRepository,
+                 jobStatusRepo: JobStatusRepository,
                  actorSystem: ActorSystem,
                  scheduler: Scheduler,
                  enableJobScheduling: Boolean) {
 
-  def this(jobs: Seq[Job], lockRepo: LockRepository, actorSystem: ActorSystem, sched: Scheduler = JobManager.createScheduler, enableJobScheduling: Boolean = true) =
-    this(Jobs(jobs), lockRepo, actorSystem, sched, enableJobScheduling)
+  def this(jobs: Seq[Job], lockRepo: LockRepository, jobStatusRepo: JobStatusRepository, actorSystem: ActorSystem, sched: Scheduler = JobManager.createScheduler, enableJobScheduling: Boolean = true) =
+    this(Jobs(jobs), lockRepo, jobStatusRepo, actorSystem, sched, enableJobScheduling)
 
   init()
 
@@ -151,9 +152,10 @@ class JobManager(managedJobs: => Jobs,
 
   def getJob(jobType: JobType): Job = managedJobs(jobType)
 
-  def allJobStatus(jobType: JobType): Future[List[JobStatus]] = managedJobs.get(jobType).get.allJobStatus
+  def allJobStatus(jobType: JobType): Future[List[JobStatus]] = jobStatusRepo.list(jobType)
+
+  def jobStatus(jobType: JobType, jobId: UUID): Future[Option[JobStatus]] = jobStatusRepo.get(jobType, jobId)
 
-  def jobStatus(jobType: JobType, jobId: UUID): Future[Option[JobStatus]] = managedJobs.get(jobType).get.jobStatus(jobId)
 }
 
 object JobManager {

src/main/scala/de/kaufhof/hajobs/JobManagerModule.scala

@@ -22,6 +22,7 @@ abstract class JobManagerModule extends Module {
   // Non-lazy because the job manager should start cron schedules
   binding toNonLazy new JobManager(jobMap,
     inject[LockRepository],
+    inject[JobStatusRepository],
     actorSystem,
     scheduler,
     inject[Configuration].getBoolean("job.import.enableJobTriggering").getOrElse(true)

src/main/scala/de/kaufhof/hajobs/JobSupervisor.scala

@@ -16,7 +16,7 @@ import scala.util.{Failure, Success}
 class JobSupervisor(jobManager: => JobManager,
                     jobUpdater: JobUpdater,
                     jobStatusRepository: JobStatusRepository,
-                    cronExpression: Option[String]) extends Job(JobTypes.JobSupervisor, jobStatusRepository, 0, cronExpression) {
+                    cronExpression: Option[String]) extends Job(JobTypes.JobSupervisor, 0, cronExpression) {
 
   def this(jobManager: => JobManager,
            lockRepository: LockRepository,

src/main/scala/de/kaufhof/hajobs/WriteStatus.scala

@@ -0,0 +1,22 @@
+package de.kaufhof.hajobs
+
+import JobState.JobState
+import org.joda.time.DateTime
+import play.api.libs.json.JsValue
+
+import scala.concurrent.Future
+
+/**
+ * Supports shortcut to store the job status, can be mixed into Jobs.
+ */
+trait WriteStatus {
+
+  def jobStatusRepository: JobStatusRepository
+  def jobType: JobType
+
+  protected def writeStatus(jobState: JobState, content: Option[JsValue] = None)(implicit jobContext: JobContext): Future[JobStatus] = {
+    val status = JobStatus(jobContext.triggerId, jobType, jobContext.jobId, jobState, JobStatus.stateToResult(jobState), DateTime.now(), content)
+    jobStatusRepository.save(status)
+  }
+
+}

src/test/scala/de/kaufhof/hajobs/ActorJobSpec.scala

@@ -21,7 +21,7 @@ class ActorJobSpec extends StandardSpec with TestKitBase {
       val testProbe = new TestProbe(system)
       val jobType = new JobType("actorJob", new LockType("actorJobLock"))
       val props = Props(actor(testProbe.ref))
-      val job = new ActorJob(jobType, _ => props, system, mock[JobStatusRepository])
+      val job = new ActorJob(jobType, _ => props, system)
 
       var finishCalled = new AtomicBoolean(false)
       implicit val context = JobContext(randomUUID(), randomUUID(), () => finishCalled.set(true))

src/test/scala/de/kaufhof/hajobs/JobManagerIntegrationSpec.scala

@@ -25,7 +25,7 @@ class JobManagerIntegrationSpec extends CassandraSpec with DefaultAwaitTimeout w
       val cdl = new CountDownLatch(1)
       val mockedScheduler = mock[Scheduler]
 
-      val manager = new JobManager(Seq(new Job1(jobStatusRepository, cdl), new Job12(jobStatusRepository)), lockRepository, mock[ActorSystem], mockedScheduler, false)
+      val manager = new JobManager(Seq(new Job1(jobStatusRepository, cdl), new Job12(jobStatusRepository)), lockRepository, jobStatusRepository, mock[ActorSystem], mockedScheduler, false)
       manager.triggerJob(JobType1)
 
       eventually {
@@ -39,7 +39,7 @@ class JobManagerIntegrationSpec extends CassandraSpec with DefaultAwaitTimeout w
 }
 
 object JobManagerIntegrationSpec {
-  class Job1(jobStatusRepository: JobStatusRepository, cdl: CountDownLatch) extends Job(JobType1, jobStatusRepository, 3) {
+  class Job1(jobStatusRepository: JobStatusRepository, cdl: CountDownLatch) extends Job(JobType1, 3) {
     override def run()(implicit context: JobContext): Future[JobStartStatus] = {
       import scala.concurrent.ExecutionContext.Implicits.global
       Future {
@@ -54,7 +54,7 @@ object JobManagerIntegrationSpec {
 
   object JobType12 extends JobType("testJob12", JobType1.lockType)
 
-  class Job12(jobStatusRepository: JobStatusRepository) extends Job(JobType12, jobStatusRepository, 3) {
+  class Job12(jobStatusRepository: JobStatusRepository) extends Job(JobType12, 3) {
     override def run()(implicit context: JobContext): Future[JobStartStatus] = {
       context.finishCallback()
       Future.successful(Started(context.jobId))

src/test/scala/de/kaufhof/hajobs/JobManagerSpec.scala

@@ -51,7 +51,7 @@ class JobManagerSpec extends StandardSpec {
       when(job.cronExpression).thenReturn(Some("* * * * * ?"))
 
 
-      val manager = new JobManager(Seq(job), lockRepository, actorSystem, scheduler, true)
+      val manager = new JobManager(Seq(job), lockRepository, jobStatusRepository, actorSystem, scheduler, true)
 
       eventually(Timeout(scaled(3 seconds))) {
         verify(job, atLeastOnce()).run()(any[JobContext])
@@ -64,7 +64,7 @@ class JobManagerSpec extends StandardSpec {
       when(job.jobType).thenReturn(JobType1)
       when(job.cronExpression).thenReturn(None)
 
-      val manager = new JobManager(Seq(job), lockRepository, actorSystem, mockedScheduler, true)
+      val manager = new JobManager(Seq(job), lockRepository, jobStatusRepository, actorSystem, mockedScheduler, true)
       verify(mockedScheduler, times(1)).start()
       verifyNoMoreInteractions(mockedScheduler)
     }
@@ -76,7 +76,7 @@ class JobManagerSpec extends StandardSpec {
       when(job.cronExpression).thenReturn(Some("* * * * * ?"))
 
       val jobUpdater = new JobUpdater(lockRepository, jobStatusRepository)
-      val manager = new JobManager(Seq(job), lockRepository, actorSystem, mockedScheduler, false)
+      val manager = new JobManager(Seq(job), lockRepository, jobStatusRepository, actorSystem, mockedScheduler, false)
       verifyNoMoreInteractions(mockedScheduler)
     }
   }
@@ -85,7 +85,7 @@ class JobManagerSpec extends StandardSpec {
     "release lock after a synchronous job finished" in {
       val job = new TestJob(jobStatusRepository)
 
-      val manager = new JobManager(Seq(job), lockRepository, actorSystem, scheduler, false)
+      val manager = new JobManager(Seq(job), lockRepository, jobStatusRepository, actorSystem, scheduler, false)
       await(manager.retriggerJob(JobType1, UUIDs.timeBased()))
 
       verify(lockRepository, times(1)).acquireLock(any(), any(), any())
@@ -100,7 +100,7 @@ class JobManagerSpec extends StandardSpec {
       when(job.jobType).thenReturn(JobType1)
       when(job.run()(any())).thenReturn(Future.failed(new RuntimeException("test exception")))
 
-      val manager = new JobManager(Seq(job), lockRepository, actorSystem, mockedScheduler, false)
+      val manager = new JobManager(Seq(job), lockRepository, jobStatusRepository, actorSystem, mockedScheduler, false)
       await(manager.retriggerJob(JobType1, UUIDs.timeBased()))
       verify(lockRepository, times(1)).acquireLock(any(), any(), any())
       verify(lockRepository, times(1)).releaseLock(any(), any())
@@ -111,8 +111,7 @@ class JobManagerSpec extends StandardSpec {
 }
 
 object JobManagerSpec {
-  class TestJob(jobStatusRepository: JobStatusRepository) extends Job(JobType1,
-    jobStatusRepository, 0) {
+  class TestJob(jobStatusRepository: JobStatusRepository) extends Job(JobType1, 0) {
     override def run()(implicit context: JobContext): Future[JobStartStatus] = {
       context.finishCallback()
       Future.successful(Started(context.jobId))

src/test/scala/de/kaufhof/hajobs/JobStatusRepositorySpec.scala

@@ -4,6 +4,7 @@ import java.util.UUID
 
 import com.datastax.driver.core.utils.UUIDs
 import de.kaufhof.hajobs.testutils.CassandraSpec
+import JobState._
 import org.joda.time.DateTime
 import play.api.libs.json.Json
 
@@ -28,9 +29,9 @@ class JobStatusRepositorySpec extends CassandraSpec {
     "get a job status by id" in {
       assume(await(repo.getAllMetadata()) === List.empty)
       // given
-      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, UUIDs.timeBased(), JobState.Finished, JobResult.Success, DateTime.now, Some(Json.toJson("muhmuh")))
+      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, UUIDs.timeBased(), Finished, JobResult.Success, DateTime.now, Some(Json.toJson("muhmuh")))
       // a job status without content should also be loaded
-      val jobStatus2: JobStatus = JobStatus(anyTriggerId, type1, UUIDs.timeBased(), JobState.Finished, JobResult.Success, DateTime.now)
+      val jobStatus2: JobStatus = JobStatus(anyTriggerId, type1, UUIDs.timeBased(), Finished, JobResult.Success, DateTime.now)
 
       // when
       await(Future.sequence((Seq(repo.save(jobStatus1), repo.save(jobStatus2)))))
@@ -52,8 +53,8 @@ class JobStatusRepositorySpec extends CassandraSpec {
       // given
       val jobId1: UUID = UUIDs.timeBased()
       val jobId2: UUID = UUIDs.timeBased()
-      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, jobId1, JobState.Finished, JobResult.Failed, DateTime.now, Some(Json.toJson("muhmuh")))
-      val jobStatus2: JobStatus = JobStatus(anyTriggerId, type1, jobId2, JobState.Finished, JobResult.Failed, DateTime.now, Some(Json.toJson("muhmuh")))
+      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, jobId1, Finished, JobResult.Failed, DateTime.now, Some(Json.toJson("muhmuh")))
+      val jobStatus2: JobStatus = JobStatus(anyTriggerId, type1, jobId2, Finished, JobResult.Failed, DateTime.now, Some(Json.toJson("muhmuh")))
 
       // when
       await(Future.sequence(Seq(repo.save(jobStatus1), repo.save(jobStatus2))))
@@ -68,8 +69,8 @@ class JobStatusRepositorySpec extends CassandraSpec {
       assume(await(repo.getAllMetadata()) === List.empty)
       val jobId1: UUID = UUIDs.timeBased()
       val jobId2: UUID = UUIDs.timeBased()
-      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, jobId1, JobState.Running, JobResult.Pending, DateTime.now, Some(Json.toJson("muhmuh")))
-      val jobStatus2: JobStatus = JobStatus(anyTriggerId, type2, jobId2, JobState.Running, JobResult.Pending, DateTime.now, Some(Json.toJson("muhmuh")))
+      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, jobId1, Running, JobResult.Pending, DateTime.now, Some(Json.toJson("muhmuh")))
+      val jobStatus2: JobStatus = JobStatus(anyTriggerId, type2, jobId2, Running, JobResult.Pending, DateTime.now, Some(Json.toJson("muhmuh")))
 
       // when
       await(Future.sequence(Seq(repo.save(jobStatus1), repo.save(jobStatus2))))
@@ -84,21 +85,21 @@ class JobStatusRepositorySpec extends CassandraSpec {
     "update meta data and insert data on update" in {
       assume(await(repo.getAllMetadata()) === List.empty)
       val jobId1: UUID = UUIDs.timeBased()
-      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, jobId1, JobState.Finished, JobResult.Failed, DateTime.now, Some(Json.toJson("muhmuh")))
+      val jobStatus1: JobStatus = JobStatus(anyTriggerId, type1, jobId1, Finished, JobResult.Failed, DateTime.now, Some(Json.toJson("muhmuh")))
 
       // when
       await(repo.save(jobStatus1))
       eventually{
         await(repo.getAllMetadata()).find(_.jobId == jobId1).map(_.jobState).value should be (jobStatus1.jobState)
       }
 
-      val jobStatus2 = await(repo.updateJobState(jobStatus1, JobState.Canceled))
+      val jobStatus2 = await(repo.updateJobState(jobStatus1, Canceled))
       eventually{
         await(repo.getAllMetadata()).find(_.jobId == jobId1).map(_.jobState).value should be (jobStatus2.jobState)
       }
 
       eventually {
-        await(repo.getJobHistory(type1, jobId1)).map(_.jobState) should contain theSameElementsAs (Seq(JobState.Canceled, JobState.Finished))
+        await(repo.getJobHistory(type1, jobId1)).map(_.jobState) should contain theSameElementsAs (Seq(Canceled, Finished))
       }
     }
   }