I'm new to akka and would like to get some feedback on my first attempt. The code takes text from stdin and finds the top 10 most frequently occurring words.
Specifically, I would like to know
- Is passing the
wordFrequencyAccumulator
to the tokenizer a reasonable thing to do? Are there better alternatives? - Is the use of
postStop
in theWordFrequencyAccumulator
to send the final result to the parent a reasonable thing to do? Are there better alternatives? - Anything else I might be doing that is weird or there is a better way?
import akka.actor._
import akka.routing._
import scala.io.Source
import scalaz._
import Scalaz._
import Utilities._
object WordCount {
case class WordFrequencies(freq: Map[Token, Int])
}
class WordCount extends Actor {
val wordFrequencyAccumulator = context.actorOf(Props[WordFrequencyAccumulator])
val tokenizeProps = Props(new Tokenize(wordFrequencyAccumulator))
val tokenizer = context.actorOf(RoundRobinPool(5).props(tokenizeProps))
context.watch(tokenizer)
Source.stdin.getLines.foreach(tokenizer ! _)
tokenizer ! Broadcast(PoisonPill)
def receive = {
case WordCount.WordFrequencies(freq) =>
freq.toList
.sortWith(_._2 > _._2)
.take(10)
.foreach(println)
context.stop(self)
case Terminated(`tokenizer`) =>
wordFrequencyAccumulator ! PoisonPill
}
}
class Tokenize(wordFrequencyAccumulator: ActorRef) extends Actor {
def receive = {
case line: String =>
val tokens = tokenize(line)
val filteredTokens = tokens
.filterNot(token => token.size < 3 || token.size > 45)
.filterNot(stopWords)
wordFrequencyAccumulator ! WordCount.WordFrequencies(frequencies(filteredTokens))
}
}
class WordFrequencyAccumulator extends Actor {
var frequencyAcc = Map[Token, Int]()
def receive = {
case WordCount.WordFrequencies(freq) => frequencyAcc = frequencyAcc |+| freq
}
override def postStop() = context.parent ! WordCount.WordFrequencies(frequencyAcc)
}