Adding rejected state to invalid streams.

build.sbt

@@ -15,7 +15,9 @@ scalacOptions += "-feature"
 libraryDependencies ++= {
   val akkaStreamV = "1.0-M3"
   val sprayV = "1.3.3"
+  val fingerV = "1.5.2+"
   Seq(
+    "de.sciss" %% "fingertree" % fingerV,
     "org.fusesource"         % "sigar" % "1.6.4",
     "io.spray"               %%  "spray-can"     % sprayV,
     "io.spray"               %%  "spray-client"     % sprayV,

src/main/resources/application.conf

@@ -7,8 +7,8 @@ akka {
 
   log-dead-letters = 1
   log-dead-letters-during-shutdown = on
-  loggers = ["akka.event.slf4j.Slf4jLogger"]
-  loglevel = "DEBUG"
+  # loggers = ["akka.event.slf4j.Slf4jLogger"]
+  # loglevel = "DEBUG"
 
   cluster {
     seed-nodes = [

src/main/scala/worker/Main.scala

@@ -53,7 +53,6 @@ object UpcloseService extends Protocols {
   implicit val system = ActorSystem()
   implicit val executor = system.dispatcher
 
-  // def config: Config
   val logger = Logging(system, getClass)
 
   lazy val config = ConfigFactory.load()

src/main/scala/worker/Master.scala

@@ -102,6 +102,19 @@ class Master(workTimeout: FiniteDuration) extends PersistentActor with ActorLogg
         }
       }
 
+    case MasterWorkerProtocol.WorkRejected(workerId, workId) =>
+      if (workState.isDone(workId)) {
+        sender() ! MasterWorkerProtocol.Ack(workId)
+      } else if (!workState.isInProgress(workId)) {
+        log.info("Work {} not in progress, reported as done by worker {}", workId, workerId)
+      } else {
+        log.info("Work {} is done by worker {}", workId, workerId)
+        changeWorkerToIdle(workerId, workId)
+        persist(WorkRejected(workId)) { event ⇒
+          workState = workState.updated(event)
+          sender ! MasterWorkerProtocol.Ack(workId)
+        }
+      }
     case MasterWorkerProtocol.WorkFailed(workerId, workId) =>
       if (workState.isInProgress(workId)) {
         log.info("Work {} failed by worker {}", workId, workerId)

src/main/scala/worker/MasterWorkerProtocol.scala

@@ -8,6 +8,7 @@ object MasterWorkerProtocol {
   case class WorkerRequestsWork(workerId: String)
   case class WorkIsDone(workerId: String, workId: String, result: EncodedVideo)
   case class WorkFailed(workerId: String, workId: String)
+  case class WorkRejected(workerId: String, workId: String)
 
   // Master -> Workers
   case object WorkIsReady

src/main/scala/worker/WorkExecutor.scala

@@ -10,7 +10,7 @@ import com.typesafe.config.ConfigFactory
 import upkoder.models.FFProbeProtocols._
 import upkoder.models._
 import upkoder.upclose.models.UpcloseBroadcast
-
+import scala.language.postfixOps
 
 
 class WorkExecutor extends Actor with ActorLogging{
@@ -26,8 +26,11 @@ class WorkExecutor extends Actor with ActorLogging{
       val bucket = "upclose-dev-thumbnails"
       log.info("Downloading {}", url)
       val srcMedia = downloadMedia(url)
+      val duration = getDuration(srcMedia.getPath)
+      log.info("Media duration {}", duration)
+      if (duration <= 1) { sender ! Worker.WorkerRejected(upcloseBroadcast.id) }
       log.info("Generating thumbnails {}", srcMedia.getPath)
-      val thumbnails = generateThumbnails(srcMedia, upcloseBroadcast.duration)
+      val thumbnails = generateThumbnails(srcMedia, duration)
       log.info("Generated thumbnails {}", srcMedia.getPath)
       val thumbsInfo = thumbnails map { x => getMediaInfo(x).transformToEncodeMedia(x, uploadToS3(x, bucket)) }
       log.info("thumbnails {}", thumbsInfo)
@@ -74,7 +77,7 @@ class WorkExecutor extends Actor with ActorLogging{
   }
 
   def generateThumbnails(srcMedia: File, duration: Int): Seq[File] = {
-    Seq.fill(3)(nextInt(getDuration(srcMedia.getPath))).map{ generateThumbnail(srcMedia.getPath, _) }
+    Seq.fill(4)(nextInt(duration)) map { generateThumbnail(srcMedia.getPath, _) }
   }
 
   def downloadMedia(url: String): File = {

src/main/scala/worker/WorkState.scala

@@ -10,11 +10,13 @@ object WorkState {
     pendingWork = PriorityQueue(),
     workInProgress = Map.empty,
     acceptedWorkIds = Set.empty,
+    rejectedWorkIds = Set.empty,
     doneWorkIds = Set.empty)
 
   trait WorkDomainEvent
   case class WorkAccepted(work: Work) extends WorkDomainEvent
   case class WorkStarted(workId: String) extends WorkDomainEvent
+  case class WorkRejected(workId: String) extends WorkDomainEvent
   case class WorkCompleted(workId: String, result: EncodedVideo) extends WorkDomainEvent
   case class WorkerFailed(workId: String) extends WorkDomainEvent
   case class WorkerTimedOut(workId: String) extends WorkDomainEvent
@@ -24,6 +26,7 @@ case class WorkState private (
   private val pendingWork: PriorityQueue[Work],
   private val workInProgress: Map[String, Work],
   private val acceptedWorkIds: Set[String],
+  private val rejectedWorkIds: Set[String],
   private val doneWorkIds: Set[String]) {
 
   import WorkState._
@@ -48,6 +51,10 @@ case class WorkState private (
       copy(
         workInProgress = workInProgress - workId,
         doneWorkIds = doneWorkIds + workId)
+    case WorkRejected(workId) =>
+      copy(
+        workInProgress = workInProgress - workId,
+        rejectedWorkIds = rejectedWorkIds + workId)
     case WorkerFailed(workId) =>
       pendingWork enqueue workInProgress(workId)
       copy(

src/main/scala/worker/Worker.scala

@@ -23,6 +23,7 @@ object Worker {
     Props(classOf[Worker], clusterClient, workExecutorProps, registerInterval)
 
   case class WorkComplete(result: EncodedVideo)
+  case class WorkerRejected(workId: Int)
 }
 
 class Worker(clusterClient: ActorRef, workExecutorProps: Props, registerInterval: FiniteDuration)
@@ -71,6 +72,16 @@ class Worker(clusterClient: ActorRef, workExecutorProps: Props, registerInterval
       context.become(working)
   }
 
+  def waitForWorkRejectedAck() : Receive = {
+    case Ack(id) if id == workId =>
+      sendToMaster(WorkerRequestsWork(workerId))
+      context.setReceiveTimeout(Duration.Undefined)
+      context.become(idle)
+    case ReceiveTimeout =>
+      log.info("No ack from master, retrying")
+      sendToMaster(WorkRejected(workerId, workId))
+  }
+
 
   def waitForWorkIsDoneAck(result: EncodedVideo): Receive = {
     case Ack(id) if id == workId =>
@@ -87,13 +98,16 @@ class Worker(clusterClient: ActorRef, workExecutorProps: Props, registerInterval
     clusterClient ! SendToAll("/user/master/active", msg)
   }
 
-
   def working: Receive = {
     case WorkComplete(result) =>
       log.info("Work is complete. Result {}.", result)
       sendToMaster(WorkIsDone(workerId, workId, result))
       context.setReceiveTimeout(5.seconds)
       context.become(waitForWorkIsDoneAck(result))
+    case WorkerRejected(workId) =>
+      sendToMaster(WorkRejected(workerId, workId.toString))
+      context.setReceiveTimeout(5.seconds)
+      context.become(waitForWorkRejectedAck())
     case _: Work =>
       log.info("Worker {} is already working", workerId)
   }