add akka executor

Author: cosminseceleanu

build.sbt

@@ -3,4 +3,8 @@ name := "PipesAndFilters"
 version := "0.1"
 
 scalaVersion := "2.12.4"
-        
+
+libraryDependencies ++= Seq(
+  "com.typesafe.akka" %% "akka-actor" % "2.5.11",
+  "com.typesafe.akka" %% "akka-testkit" % "2.5.11" % Test
+)

src/main/scala/com/cosmin/examples/AkkaPipeline.scala

@@ -0,0 +1,18 @@
+package com.cosmin.examples
+
+import com.cosmin.pipeline.executor.AkkaExecutor
+
+import scala.util.Success
+
+object AkkaPipeline {
+  def main(args: Array[String]): Unit = {
+    val exec = AkkaExecutor[Int, Int]
+//    exec.execute(1, List()) {
+//      case Success(nr) => println("aaaa")
+//    }
+
+    val l = List(1, 2, 3)
+    println(l.head)
+    println(l.tail)
+  }
+}

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

@@ -1,7 +1,7 @@
 package com.cosmin.examples.wordcount
 
 import com.cosmin.pipeline.Pipeline
-import com.cosmin.pipeline.executor.AsyncExecutor
+import com.cosmin.pipeline.executor.{AkkaExecutor, AsyncExecutor}
 
 import scala.util.Success
 
@@ -18,6 +18,7 @@ object WordCount {
     }
 
     executeAsync(pipeline, wordToFind)
+    executeAsyncUsingAkka(pipeline, wordToFind)
   }
 
   private def executeAsync(pipeline: Pipeline[String, Int], wordToFind: String): Unit = {
@@ -27,4 +28,11 @@ object WordCount {
     }
     Thread.sleep(2000)
   }
+
+  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")
+    }
+  }
 }

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

@@ -0,0 +1,20 @@
+package com.cosmin.pipeline.executor
+import akka.actor.ActorSystem
+import com.cosmin.pipeline.Stage
+import com.cosmin.pipeline.executor.actor.Supervisor
+import com.cosmin.pipeline.executor.actor.Supervisor.Start
+
+import scala.util.Try
+
+class AkkaExecutor[In, Out] extends PipelineExecutor[In, Out] {
+  override def execute(in: In, stages: List[Stage])(onComplete: Try[Out] => Unit): Unit = {
+    val system = ActorSystem("pipeline")
+    val supervisor = system.actorOf(Supervisor.props[Out](stages, onComplete), "pipeline-supervisor")
+
+    supervisor ! Start[In](in)
+  }
+}
+
+object AkkaExecutor {
+  def apply[In, Out]: AkkaExecutor[In, Out] = new AkkaExecutor[In, Out]
+}

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

@@ -0,0 +1,39 @@
+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.Worker.Execute
+
+import scala.util.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)
+}
+
+class Supervisor[Out](stages: List[Stage], onComplete: Try[Out] => Unit) extends Actor with ActorLogging {
+  private var remainingStages = stages
+  private var stageToActor: Map[Stage, ActorRef] = Map.empty
+
+  override def preStart(): Unit = stageToActor = stages.map(stage => (stage, context.actorOf(Worker.props(stage)))).toMap
+
+  override def receive: Receive = {
+    case Start(in) => executeStage(in)
+    case StageCompleted(stage, result) =>
+      if (remainingStages.isEmpty) {
+        onComplete(Try(result.asInstanceOf[Out]))
+        context.stop(self)
+      } else {
+        executeStage(result)
+      }
+  }
+
+  private def executeStage(in: Any): Unit = {
+    val stage = remainingStages.head
+    stageToActor(stage) ! Execute[stage.In](in.asInstanceOf[stage.In])
+    remainingStages = remainingStages.tail
+  }
+}

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

@@ -0,0 +1,20 @@
+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.Worker.Execute
+
+object Worker {
+  def props(stage: Stage): Props = Props(new Worker(stage))
+
+  final case class Execute[In](in: In)
+}
+
+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])
+  }
+}