fix error handling + word count example with failure

Author: cosminseceleanu

src/main/scala/com/cosmin/examples/wordcount/WordCount.scala

@@ -1,9 +1,9 @@
 package com.cosmin.examples.wordcount
 
 import com.cosmin.pipeline.Pipeline
-import com.cosmin.pipeline.executor.{AkkaExecutor, AsyncExecutor}
+import com.cosmin.pipeline.executor.{AkkaExecutor, AsyncExecutor, PipelineExecutor, SynchronouslyExecutor}
 
-import scala.util.Success
+import scala.util.{Failure, Success}
 
 /**
   * Implementation of the following UNIX command 'cat "myText.txt" | grep "hello" | wc -l'
@@ -13,26 +13,29 @@ object WordCount {
     val wordToFind = "hello"
     val pipeline = Pipeline[String, String]() | Cat() | Grep(wordToFind) | Count()
 
-    pipeline.execute("myText.txt") {
-      case Success(output) => println(s"word '$wordToFind' was found on $output lines")
-    }
+    doExecute(pipeline, SynchronouslyExecutor(), "Sync", wordToFind)
+    doExecute(pipeline, SynchronouslyExecutor(), "Sync Failed", wordToFind) ("notFound.txt")
 
     executeAsync(pipeline, wordToFind)
     executeAsyncUsingAkka(pipeline, wordToFind)
   }
 
   private def executeAsync(pipeline: Pipeline[String, Int], wordToFind: String): Unit = {
-    implicit val asyncExecutor: AsyncExecutor[String, Int] = AsyncExecutor[String, Int]
-    pipeline.execute("myText.txt") {
-      case Success(output) => println(s"Async ---> word '$wordToFind' was found on $output lines")
-    }
-    Thread.sleep(2000)
+    val asyncExecutor: AsyncExecutor[String, Int] = AsyncExecutor[String, Int]
+    doExecute(pipeline, asyncExecutor, "Async", wordToFind)
+    doExecute(pipeline, asyncExecutor, "Async Failed", wordToFind) ("notFound.txt")
   }
 
   private def executeAsyncUsingAkka(pipeline: Pipeline[String, Int], wordToFind: String): Unit = {
-    implicit val akkaExecutor: AkkaExecutor[String, Int] = AkkaExecutor[String, Int]
-    pipeline.execute("myText.txt") {
-      case Success(output) => println(s"Akka Async ---> word '$wordToFind' was found on $output lines")
-    }
+    val akkaExecutor: AkkaExecutor[String, Int] = AkkaExecutor[String, Int]
+    doExecute(pipeline, akkaExecutor, "Akka Async", wordToFind)
+    doExecute(pipeline, akkaExecutor, "Akka Async Failed", wordToFind) ("notFound.txt")
+  }
+
+  private def doExecute(pipe: Pipeline[String, Int], exec: PipelineExecutor[String, Int], prefix: String, word: String) (implicit file: String = "myText.txt") : Unit = {
+    pipe.execute(file) {
+      case Success(output) => println(s"$prefix ---> word '$word' was found on $output lines")
+      case Failure(e) => println(s"$prefix ---> error: $e")
+    } (exec)
   }
 }

src/main/scala/com/cosmin/pipeline/executor/AkkaExecutor.scala

@@ -8,7 +8,7 @@ import scala.util.Try
 
 class AkkaExecutor[In, Out](system: ActorSystem) extends PipelineExecutor[In, Out] {
   override def execute(in: In, stages: List[Stage])(onComplete: Try[Out] => Unit): Unit = {
-    val supervisor = system.actorOf(Supervisor.props[Out](stages, onComplete), "pipeline-supervisor")
+    val supervisor = system.actorOf(Supervisor.props[Out](stages, onComplete))
 
     supervisor ! Start[In](in)
   }

src/main/scala/com/cosmin/pipeline/executor/SynchronouslyExecutor.scala

@@ -6,9 +6,9 @@ import scala.util.Try
 
 class SynchronouslyExecutor[In, Out] extends PipelineExecutor[In, Out] {
   override def execute(in: In, stages: List[Stage])(onComplete: Try[Out] => Unit): Unit = {
-    val result: Out = doExecute[In, Out](stages, in)
+    val result: Try[Out] = Try[Out](doExecute[In, Out](stages, in))
 
-    onComplete(Try[Out](result))
+    onComplete(result)
   }
 
   def doExecute[I, O](stages: List[Stage], input: I): O = stages match {

src/main/scala/com/cosmin/pipeline/executor/actor/Supervisor.scala

@@ -2,16 +2,17 @@ package com.cosmin.pipeline.executor.actor
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Props}
 import com.cosmin.pipeline.Stage
-import com.cosmin.pipeline.executor.actor.Supervisor.{StageCompleted, Start}
+import com.cosmin.pipeline.executor.actor.Supervisor.{StageCompleted, StageFailed, Start}
 import com.cosmin.pipeline.executor.actor.Worker.Execute
 
-import scala.util.Try
+import scala.util.{Failure, Try}
 
 object Supervisor {
   def props[Out](stages: List[Stage], onComplete: Try[Out] => Unit): Props = Props(new Supervisor(stages, onComplete))
 
   final case class Start[In](in: In)
   final case class StageCompleted[Out](stage: Stage, result: Out)
+  final case class StageFailed(stage: Stage, e: Throwable)
 }
 
 class Supervisor[Out](stages: List[Stage], onComplete: Try[Out] => Unit) extends Actor with ActorLogging {
@@ -22,13 +23,19 @@ class Supervisor[Out](stages: List[Stage], onComplete: Try[Out] => Unit) extends
 
   override def receive: Receive = {
     case Start(in) => executeStage(in)
-    case StageCompleted(stage, result) =>
-      if (remainingStages.isEmpty) {
+    case StageCompleted(stage, result) => onStageCompleted(result)
+    case StageFailed(stage, e) =>
+      onComplete(Failure(e))
+      context.stop(self)
+  }
+
+  private def onStageCompleted(result: Any): Unit = {
+    remainingStages match {
+      case Nil =>
         onComplete(Try(result.asInstanceOf[Out]))
         context.stop(self)
-      } else {
-        executeStage(result)
-      }
+      case _ => executeStage(result)
+    }
   }
 
   private def executeStage(in: Any): Unit = {

src/main/scala/com/cosmin/pipeline/executor/actor/Worker.scala

@@ -2,9 +2,11 @@ package com.cosmin.pipeline.executor.actor
 
 import akka.actor.{Actor, Props}
 import com.cosmin.pipeline.Stage
-import com.cosmin.pipeline.executor.actor.Supervisor.StageCompleted
+import com.cosmin.pipeline.executor.actor.Supervisor.{StageCompleted, StageFailed}
 import com.cosmin.pipeline.executor.actor.Worker.Execute
 
+import scala.util.{Failure, Success, Try}
+
 object Worker {
   def props(stage: Stage): Props = Props(new Worker(stage))
 
@@ -14,7 +16,13 @@ object Worker {
 class Worker(stage: Stage) extends Actor {
   override def receive: Receive = {
     case Execute(input) =>
-      val result = stage.execute(input.asInstanceOf[stage.In])
-      sender() ! StageCompleted[stage.Out](stage, result.asInstanceOf[stage.Out])
+       executeStage(input)
+  }
+
+  private def executeStage(input: Any): Unit = {
+    Try[stage.Out](stage.execute(input.asInstanceOf[stage.In])) match {
+      case Success(result) => sender() ! StageCompleted[stage.Out](stage, result.asInstanceOf[stage.Out])
+      case Failure(e) => sender() ! StageFailed(stage, e)
+    }
   }
 }