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, Behavior, Scheduler}
24 import akka.actor.typed.scaladsl.AskPattern._
25 import akka.actor.typed.scaladsl.Behaviors
26 import akka.util.Timeout
27 import grizzled.slf4j.Logging
28 import org.make.api.technical.ActorProtocol
29 import org.make.api.technical.job.JobReportingActor.Protocol.{Command, Response}
30 import org.make.core.job.Job.{JobId, Progress}
31 
32 import scala.concurrent.ExecutionContext.Implicits.global
33 import scala.concurrent.Future
34 import scala.concurrent.duration.FiniteDuration
35 import scala.util.{Failure, Success}
36 
37 object JobReportingActor extends Logging {
38 
39   final class JobReportingActorFacade private (actor: ActorRef[Command]) {
40     def apply(progress: Progress)(implicit timeout: Timeout, scheduler: Scheduler): Future[Unit] =
41       (actor ? (Command.Report(progress, _))).void
42   }
43 
44   object JobReportingActorFacade {
45     def apply(actor: ActorRef[Command]): JobReportingActorFacade = new JobReportingActorFacade(actor)
46   }
47 
48   sealed abstract class Protocol extends ActorProtocol
49 
50   object Protocol {
51     sealed abstract class Command extends Protocol
52 
53     object Command {
54       final case class Report(progress: Progress, replyTo: ActorRef[Response.Ack.type]) extends Command
55       final case class Finish(outcome: Option[Throwable]) extends Command
56       case object Tick extends Command
57       case object HeartbeatSuccess extends Command
58       case object HeartbeatFailure extends Command
59       final case class ReportResult(replyTo: ActorRef[Response.Ack.type]) extends Command
60       case object Stop extends Command
61     }
62 
63     sealed abstract class Response extends Protocol
64 
65     object Response {
66       case object Ack extends Response
67     }
68 
69   }
70 
71   def apply(
72     jobId: JobId,
73     work: JobReportingActorFacade => Future[Unit],
74     jobCoordinatorService: JobCoordinatorService,
75     heartRate: FiniteDuration
76   ): Behavior[Protocol.Command] = {
77     Behaviors.setup { context =>
78       Behaviors.withTimers { timers =>
79         timers.startTimerWithFixedDelay(s"${context.self.path.name}-heartbeat", Command.Tick, heartRate)
80 
81         context.pipeToSelf(work(JobReportingActorFacade(context.self))) { result =>
82           Command.Finish(result.failed.toOption)
83         }
84 
85         Behaviors.receiveMessage {
86           case Command.Tick =>
87             val futureResult = jobCoordinatorService.heartbeat(jobId)
88             context.pipeToSelf(futureResult) {
89               case Success(_) => Command.HeartbeatSuccess
90               case Failure(e) =>
91                 logger.error(s"Could not send heartbeat for job $jobId", e)
92                 Command.HeartbeatFailure
93             }
94             Behaviors.same
95           case Command.HeartbeatSuccess => Behaviors.same
96           case Command.HeartbeatFailure => Behaviors.stopped
97           case Command.Report(progress, replyTo) =>
98             val futureResult = jobCoordinatorService.report(jobId, progress)
99             context.pipeToSelf(futureResult) {
100               case Success(_) => Command.ReportResult(replyTo)
101               case Failure(e) =>
102                 logger.error(s"Job $jobId failed to report", e)
103                 Command.ReportResult(replyTo)
104             }
105             Behaviors.same
106           case Command.ReportResult(replyTo) =>
107             replyTo ! Response.Ack
108             Behaviors.same
109           case Command.Finish(outcome) =>
110             val futureResult = jobCoordinatorService.finish(jobId, outcome)
111             context.pipeToSelf(futureResult) {
112               case Success(_) => Command.Stop
113               case Failure(e) =>
114                 logger.error(s"Job $jobId failed to finish", e)
115                 Command.Stop
116             }
117             Behaviors.same
118           case Command.Stop => Behaviors.stopped
119         }
120       }
121     }
122 
123   }
124 }