jobActor.scala

import akka.actor._
import akka.pattern.{after, ask, pipe}
import akka.util.Timeout

import scala.concurrent.ExecutionContext.Implicits._
import scala.concurrent._
import scala.concurrent.duration._

class Job(jobTracker: ActorRef) extends Actor {
  var count = 0

  def perform { 
    count += 1
    println("doing work")
  } 
  
  def receive = {
    case JobRequestMessage =>
      perform
      jobTracker ! JobDoneMessage
      
    case JobStatusMessage =>
      val c = count
      jobTracker ! JobStatusMessage(s"count is $c")
      
    case JobStopMessage =>
      context.stop(self)
      
    case JobReq(worktimes, timeout, async) => {
      val searchFutures = worktimes map { worktime =>
        doWork(worktime)
        val fallback = after(timeout, context.system.scheduler) {
          Future successful s"$worktime ms > $timeout"
        }
        Future firstCompletedOf Seq(searchFuture, fallback)
      }

      // Pipe future results to sender
      (Future sequence searchFutures) pipeTo sender
    }
  }
  
  class JobTracker extends Actor {
    def receive = {
      case JobStatusMessage(m) =>
        println(s"job status: $m")
        
      case JobDoneMessage =>
        println("--job done")
        sender ! JobStopMessage
        context.stop(self)
    }
  }
  
  object JobTest {
    // extends App
    def run = {
      println("running")
      val system = ActorSystem("JobSystem")
      val jobTracker = system.actorOf(Props[JobTracker], name = "jobTracker")
      val job = system.actorOf(Props(new Job(jobTracker)), name = "job")
      job ! JobRequestMessage
    }
  }