Added new Akka example

Author: jsuereth

chapter9/build.sbt

@@ -0,0 +1,9 @@
+name := "scatter-gather"
+
+organization := "scalaindepth"
+
+resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases"
+
+libraryDependencies += "se.scalablesolutions.akka" % "akka-actor" % "1.1.2"
+
+

chapter9/project/build.properties

@@ -0,0 +1,91 @@
+package scattergather
+
+import collection.immutable.HashMap
+import akka.actor.{ReceiveTimeout, ActorRef, Actor}
+
+/**
+ * A message representing a document to add to the search tree.
+ */
+case class SearchableDocument(content: String)
+
+trait BaseHeadNode { self: AdaptiveSearchNode =>
+ var children = IndexedSeq[ActorRef]()
+ var currentIdx = 0
+ def parentNode: PartialFunction[Any, Unit] = {
+    case SearchQuery(q, max, responder) =>
+        // TODO - use gatherer scheudler
+        val gatherer = Actor.actorOf(new GathererNode {
+          val maxDocs = max
+          val maxResponses = children.size
+          val query = q
+          val client = responder
+        })
+        gatherer.start
+        for (node <- children) {
+          node ! SearchQuery(q, max, gatherer)
+        }
+    case s @ SearchableDocument(_) => getNextChild ! s
+  }
+
+  // Round Robin
+  private def getNextChild = {
+    currentIdx = (1 + currentIdx) % children.size
+    children(currentIdx)
+  }
+
+}
+
+trait BaseChildNode { self: AdaptiveSearchNode =>
+  final val maxNoOfDocuments = 10
+  var documents: Vector[String] = Vector()
+  var index: HashMap[String, Seq[(Double, String)]] = HashMap()
+
+  def leafNode: PartialFunction[Any, Unit] = {
+    case SearchQuery(query, maxDocs, handler) => executeLocalQuery(query, maxDocs, handler)
+    case SearchableDocument(content)          => addDocumentToLocalIndex(content)
+  }
+  private def executeLocalQuery(query: String, maxDocs: Int, handler: ActorRef) = {
+    val result = for {
+      results <- index.get(query).toList
+      resultList <- results
+    } yield resultList
+    handler ! QueryResponse(result take maxDocs)
+  }
+
+  private def addDocumentToLocalIndex(content: String) = {
+    for( (key,value) <- content.split("\\s+").groupBy(identity)) {
+      val list = index.get(key) getOrElse Seq()
+      index += ((key, ((value.length.toDouble, content)) +: list))
+    }
+    documents = documents :+ content
+    // Split on size....
+    if (documents.size > maxNoOfDocuments) split()
+  }
+
+  /** Abstract method to split this actor. */
+  protected def split(): Unit
+
+  protected def clearIndex(): Unit = {
+    documents = Vector()
+    index = HashMap()
+  }
+}
+
+
+class AdaptiveSearchNode extends Actor with BaseHeadNode with BaseChildNode {
+  def receive = leafNode
+
+  /** Splits this search node into a tree of search nodes if there are too many documents. */
+  protected def split(): Unit = {
+    children = (for(docs <- documents grouped 5) yield {
+      // TODO - use search scheduler + hook up to supervisor...
+      val child = Actor.actorOf[AdaptiveSearchNode]
+      child.start()
+      docs foreach (child ! SearchableDocument(_))
+      child
+    }).toIndexedSeq
+    clearIndex()
+    this become parentNode
+  }
+
+}

chapter9/project/build/chapter9.scala

@@ -1,10 +1,43 @@
 package scattergather
 
-import akka.actor.{Supervisor, Actor}
+import akka.actor.{Supervisor, Actor, ActorRef}
 import akka.config.Supervision._
 import akka.dispatch.Dispatchers
 import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
 
+object AdaptiveSearchTreeMain {
+  def submitInitialDocuments(searchNode: ActorRef) =
+    Seq("Some example data for you",
+        "Some more example data for you to use",
+        "To be or not to be, that is the question",
+        "OMG it's a cat",
+        "This is an example.  It's a great one",
+        "HAI there", 
+        "HAI IZ HUNGRY",
+        "Hello, World",
+        "Hello, and welcome to the search node 8",
+        "The lazy brown fox jumped over the",
+        "Winning is the best because it's winning."
+    ) foreach (doc =>  searchNode ! SearchableDocument(doc))
+  def makeTree = {
+    val supervisor = Supervisor(SupervisorConfig(AllForOneStrategy(List(classOf[Exception]), 3, 1000), Nil))
+    val searchnodedispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher("adaptive search tree")
+        .withNewThreadPoolWithLinkedBlockingQueueWithCapacity(100)
+        .setCorePoolSize(10)
+        .setMaxPoolSize(128)
+        .setKeepAliveTimeInMillis(60000)
+        .setRejectionPolicy(new CallerRunsPolicy)
+        .build
+    val searchTree = Actor.actorOf(new AdaptiveSearchNode {
+      self.dispatcher = searchnodedispatcher
+    })
+    supervisor link searchTree
+    searchTree.start()
+    submitInitialDocuments(searchTree)
+    searchTree
+  }
+}
+
 object SearchTreeMain {
   def create(name: String) = {
     val supervisor = Supervisor(SupervisorConfig(AllForOneStrategy(List(classOf[Exception]), 3, 1000), Nil))
@@ -39,4 +72,4 @@ object SearchTreeMain {
     tmp
   }
 
-}
+}