add queue and pull patterns

Author: rkuhn

build.sbt

@@ -15,3 +15,9 @@ lazy val chapter12 = project dependsOn (common)
 lazy val chapter13 = project dependsOn (common)
 
 lazy val chapter14 = project dependsOn (common)
+
+lazy val chapter15 = project dependsOn (common)
+
+lazy val chapter16 = project dependsOn (common)
+
+lazy val chapter17 = project dependsOn (common)

chapter15/build.sbt

@@ -0,0 +1,3 @@
+import Build._
+
+libraryDependencies ++= Seq(akkaActor, akkaTyped, akkaPersistence, akkaPersistenceQuery, akkaStream, junit, scalatest, levelDb)

chapter15/src/main/scala/com/reactivedesignpatterns/chapter15/PullPattern.scala

@@ -0,0 +1,79 @@
+/**
+ * Copyright (C) 2015 Roland Kuhn <http://rolandkuhn.com>
+ */
+package com.reactivedesignpatterns.chapter15
+
+import java.math.MathContext
+import java.math.RoundingMode
+
+import akka.actor.Actor
+import akka.actor.ActorRef
+import akka.actor.ActorSystem
+import akka.actor.Props
+
+object PullPattern {
+
+  case class Job(id: Long, input: Int, replyTo: ActorRef)
+  case class Result(id: Long, report: BigDecimal)
+
+  case class WorkRequest(worker: ActorRef, items: Int)
+
+  class Manager extends Actor {
+
+    val workStream: Iterator[Job] =
+      Iterator from 1 map (x => Job(x, x, self)) take 1000000
+
+    val aggregator = (x: BigDecimal, y: BigDecimal) => x + y
+    var approximation = BigDecimal(0, new MathContext(10000, RoundingMode.HALF_EVEN))
+
+    var outstandingWork = 0
+
+    (1 to 8) foreach (_ => context.actorOf(Props(new Worker(self))))
+
+    def receive = {
+      case WorkRequest(worker, items) =>
+        workStream.take(items).foreach {
+          job =>
+            worker ! job
+            outstandingWork += 1
+        }
+      case Result(id, report) =>
+        approximation = aggregator(approximation, report)
+        outstandingWork -= 1
+        if (outstandingWork == 0 && workStream.isEmpty) {
+          println("final result: " + approximation)
+          context.system.terminate()
+        }
+    }
+  }
+
+  class Worker(manager: ActorRef) extends Actor {
+    val mc = new MathContext(100, RoundingMode.HALF_EVEN)
+    val plus = BigDecimal(1, mc)
+    val minus = BigDecimal(-1, mc)
+
+    var requested = 0
+    def request(): Unit =
+      if (requested < 5) {
+        manager ! WorkRequest(self, 10)
+        requested += 10
+      }
+
+    request()
+
+    def receive = {
+      case Job(id, data, replyTo) =>
+        requested -= 1
+        request()
+        val sign = if ((data & 1) == 1) plus else minus
+        val result = sign / data
+        replyTo ! Result(id, result)
+    }
+  }
+
+  def main(args: Array[String]): Unit = {
+    val sys = ActorSystem("pi")
+    sys.actorOf(Props(new Manager), "manager")
+  }
+
+}

chapter15/src/main/scala/com/reactivedesignpatterns/chapter15/QueuePattern.scala

@@ -0,0 +1,91 @@
+/**
+ * Copyright (C) 2015 Roland Kuhn <http://rolandkuhn.com>
+ */
+package com.reactivedesignpatterns.chapter15
+
+import java.math.MathContext
+import java.math.RoundingMode
+import akka.actor.Actor
+import akka.actor.ActorRef
+import akka.actor.ActorSystem
+import akka.actor.Props
+import akka.pattern.extended.ask
+import scala.collection.immutable.Queue
+import scala.concurrent.duration._
+import akka.util.Timeout
+import scala.concurrent.Future
+import scala.concurrent.Await
+import java.util.concurrent.TimeoutException
+
+object QueuePattern {
+
+  case class Job(id: Long, input: Int, replyTo: ActorRef)
+  case class JobRejected(id: Long)
+  case class Result(id: Long, report: BigDecimal)
+
+  case class WorkRequest(worker: ActorRef, items: Int)
+
+  class Manager extends Actor {
+
+    var workQueue = Queue.empty[Job]
+    var requestQueue = Queue.empty[ActorRef]
+
+    (1 to 8) foreach (_ => context.actorOf(Props(new Worker(self))))
+
+    def receive = {
+      case job @ Job(id, _, replyTo) =>
+        if (requestQueue.isEmpty) {
+          if (workQueue.size < 1000) workQueue :+= job
+          else replyTo ! JobRejected(id)
+        } else {
+          requestQueue.head ! job
+          requestQueue = requestQueue.drop(1)
+        }
+      case WorkRequest(worker, items) =>
+        if (workQueue.isEmpty) {
+          if (!requestQueue.contains(worker)) requestQueue :+= worker
+        } else {
+          workQueue.iterator.take(items).foreach(job => worker ! job)
+          workQueue = workQueue.drop(items)
+        }
+    }
+  }
+
+  class Worker(manager: ActorRef) extends Actor {
+    val mc = new MathContext(100, RoundingMode.HALF_EVEN)
+    val plus = BigDecimal(1, mc)
+    val minus = BigDecimal(-1, mc)
+
+    manager ! WorkRequest(self, 1)
+
+    def receive = {
+      case Job(id, data, replyTo) =>
+        manager ! WorkRequest(self, 1)
+        val sign = if ((data & 1) == 1) plus else minus
+        val result = sign / data
+        replyTo ! Result(id, result)
+    }
+  }
+
+  def main(args: Array[String]): Unit = {
+    val sys = ActorSystem("pi")
+    import sys.dispatcher
+    val calculator = sys.actorOf(Props(new Manager), "manager")
+    implicit val timeout = Timeout(10.seconds)
+    val futures =
+      (1 to 1000000).map(i =>
+        (calculator ? (Job(i, i, _)))
+          .collect {
+            case Result(_, report) => report
+            case _                 => BigDecimal(0)
+          })
+    Future.reduce(futures)(_ + _)
+      .map(x => println("final result: " + x * 4))
+      .recover {
+        case ex =>
+          ex.printStackTrace()
+      }
+      .foreach(_ => sys.terminate())
+  }
+
+}

chapter16/build.sbt

@@ -0,0 +1,3 @@
+import Build._
+
+libraryDependencies ++= Seq(akkaActor, akkaTyped, akkaPersistence, akkaPersistenceQuery, akkaStream, junit, scalatest, levelDb)

chapter17/build.sbt

@@ -0,0 +1,3 @@
+import Build._
+
+libraryDependencies ++= Seq(akkaActor, akkaTyped, akkaPersistence, akkaPersistenceQuery, akkaStream, junit, scalatest, levelDb)