1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.api.technical.job
21 
22 import cats.implicits._
23 import akka.actor.typed.ActorRef
24 import akka.actor.typed.scaladsl.AskPattern._
25 import akka.util.Timeout
26 import org.make.api.technical.job.JobActor.Protocol.Command._
27 import org.make.api.technical.job.JobActor.Protocol.Response._
28 import org.make.api.technical.job.JobReportingActor.JobReportingActorFacade
29 import org.make.api.technical.{ActorSystemComponent, IdGeneratorComponent, SpawnActorServiceComponent, TimeSettings}
30 import org.make.core.job.Job
31 import org.make.core.job.Job.{JobId, Progress}
32 
33 import scala.concurrent.ExecutionContext.Implicits.global
34 import scala.concurrent.Future
35 import scala.concurrent.duration.FiniteDuration
36 
37 trait JobCoordinatorComponent {
38   def jobCoordinator: ActorRef[JobActor.Protocol.Command]
39 }
40 
41 trait JobCoordinatorService {
42   def start(id: JobId, heartRate: FiniteDuration = Job.defaultHeartRate)(
43     work: JobReportingActorFacade => Future[Unit]
44   ): Future[JobAcceptance]
45   def heartbeat(id: JobId): Future[Unit]
46   def report(id: JobId, value: Progress): Future[Unit]
47   def finish(id: JobId, outcome: Option[Throwable]): Future[Unit]
48   def get(id: JobId): Future[Option[Job]]
49 }
50 
51 trait JobCoordinatorServiceComponent {
52   def jobCoordinatorService: JobCoordinatorService
53 }
54 
55 trait DefaultJobCoordinatorServiceComponent extends JobCoordinatorServiceComponent {
56   self: ActorSystemComponent with IdGeneratorComponent with JobCoordinatorComponent with SpawnActorServiceComponent =>
57 
58   override lazy val jobCoordinatorService: JobCoordinatorService = new DefaultJobCoordinatorService
59 
60   class DefaultJobCoordinatorService extends JobCoordinatorService {
61 
62     implicit val timeout: Timeout = TimeSettings.defaultTimeout
63 
64     override def start(id: JobId, heartRate: FiniteDuration)(
65       work: JobReportingActorFacade => Future[Unit]
66     ): Future[JobAcceptance] = {
67       (jobCoordinator ? (Start(id, _))).flatMap {
68         case acceptance @ JobAcceptance(true) =>
69           spawnActorService
70             .spawn(
71               behavior = JobReportingActor(id, work, jobCoordinatorService, heartRate),
72               name = s"JobReportingActor-${id.value}"
73             )
74             .map(_ => acceptance)
75         case acceptance => Future.successful(acceptance)
76       }
77     }
78 
79     override def heartbeat(id: JobId): Future[Unit] = {
80       (jobCoordinator ? (Heartbeat(id, _))).void
81     }
82 
83     override def report(id: JobId, progress: Progress): Future[Unit] = {
84       (jobCoordinator ? (Report(id, progress, _))).void
85     }
86 
87     override def finish(id: JobId, outcome: Option[Throwable]): Future[Unit] = {
88       (jobCoordinator ? (Finish(id, outcome, _))).void
89     }
90 
91     override def get(id: JobId): Future[Option[Job]] = {
92       (jobCoordinator ? (Get(id, _))).map(_.value)
93     }
94 
95   }
96 
97 }