title: Scala Akka Consumer/Producer: Return Value
body: <h2>Problem Statement</h2>
  
  <p>Assume I have a file with sentences that is processed line by line. In my case, I need to extract Named Entities (Persons, Organizations, ...) from these lines. Unfortunately, the tagger is quite slow. Therefore, I decided to parallelize the computation, such that lines could be processed independent from each other and the result is collected in a central location. </p>
  
  <h2>Current Approach</h2>
  
  <p>My current approach comprises the usage of a single producer multiple consumer concept. However, I'm relative new to Akka, but I think my problem description fits well into its capabilities. Let me show you some code:</p>
  
  <h3>Producer</h3>
  
  <p>The <code>Producer</code> reads the file line by line and sends it to the <code>Consumer</code>. If it reaches the total line limit, it propagates the result back to <code>WordCount</code>.</p>
  
  <pre><code>class Producer(consumers: ActorRef) extends Actor with ActorLogging {
  
    var master: Option[ActorRef] = None
  
    var result = immutable.List[String]()
    var totalLines = 0
    var linesProcessed = 0
  
  
    override def receive = {
      case StartProcessing() =&gt; {
        master = Some(sender)
  
        Source.fromFile("sent.txt", "utf-8").getLines.foreach { line =&gt;
          consumers ! Sentence(line)
          totalLines += 1
        }
  
        context.stop(self)
      }
  
      case SentenceProcessed(list) =&gt; {
        linesProcessed += 1
        result :::= list
        //If we are done, we can propagate the result to the creator
        if (linesProcessed == totalLines) {
          master.map(_ ! result)
        }
      }
  
      case _ =&gt; log.error("message not recognized")
    }
  }
  </code></pre>
  
  <h3>Consumer</h3>
  
  <pre><code>class Consumer extends Actor with ActorLogging {
  
    def tokenize(line: String): Seq[String] = {
      line.split(" ").map(_.toLowerCase)
    }
  
    override def receive = {
      case Sentence(sent) =&gt; {
        //Assume: This is representative for the extensive computation method 
        val tokens = tokenize(sent)
  
        sender() ! SentenceProcessed(tokens.toList)
    }
  
      case _ =&gt; log.error("message not recognized")
    }
  }
  </code></pre>
  
  <h3>WordCount (Master)</h3>
  
  <pre><code>class WordCount extends Actor {
  
    val consumers = context.actorOf(Props[Consumer].
      withRouter(FromConfig()).
      withDispatcher("consumer-dispatcher"), "consumers")
    val producer = context.actorOf(Props(new Producer(consumers)), "producer")
  
    context.watch(consumers)
    context.watch(producer)
  
    def receive = {
      case Terminated(`producer`) =&gt; consumers ! Broadcast(PoisonPill)
      case Terminated(`consumers`) =&gt; context.system.shutdown
    }
  }
  
  object WordCount {
  
    def getActor() = new WordCount
  
    def getConfig(routerType: String, dispatcherType: String)(numConsumers: Int) = s"""
        akka.actor.deployment {
          /WordCount/consumers {
            router = $routerType
            nr-of-instances = $numConsumers
            dispatcher = consumer-dispatcher
          }
        }
        consumer-dispatcher {
          type = $dispatcherType
          executor = "fork-join-executor"
        }"""
  }
  </code></pre>
  
  <p>The <code>WordCount</code> actor is responsible for creating the other actors. When the <code>Consumer</code> is finished the <code>Producer</code> sends a message with all tokens. But, how to propagate the message again and also accept and wait for it? The architecture with the third <code>WordCount</code> actor might be wrong.</p>
  
  <h3>Main Routine</h3>
  
  <pre><code>case class Run(name: String, actor: () =&gt; Actor, config: (Int) =&gt; String)
  
  object Main extends App {
  
    val run = Run("push_implementation", WordCount.getActor _, WordCount.getConfig("balancing-pool", "Dispatcher") _)
  
    def execute(run: Run, numConsumers: Int) = {
  
      val config = ConfigFactory.parseString(run.config(numConsumers))
  
      val system = ActorSystem("Counting", ConfigFactory.load(config))
      val startTime = System.currentTimeMillis
  
      system.actorOf(Props(run.actor()), "WordCount")
      /*
            How to get the result here?!
      */
      system.awaitTermination
  
      System.currentTimeMillis - startTime
    }
  
  
     execute(run, 4)
  }
  </code></pre>
  
  <h3>Problem</h3>
  
  <p>As you see, the actual problem is to propagate the result back to the <code>Main</code> routine. Can you tell me how to do this in a proper way? The question is also how to wait for the result until the consumers are finished? I had a brief look into the Akka <code>Future</code> documentation section, but the whole system is a little bit overwhelming for beginners. Something like <code>var future = message ? actor</code> seems suitable. Not sure, how to do this. Also using the <code>WordCount</code> actor causes additional complexity. Maybe it is possible to come up with a solution that doesn't need this actor? </p>