refine ActiveActive and start chapter13

in particular add EC2Worker

Author: rkuhn

build.sbt

@@ -11,3 +11,5 @@ lazy val chapter04 = project dependsOn (common)
 lazy val chapter07 = project dependsOn (common)
 
 lazy val chapter12 = project dependsOn (common)
+
+lazy val chapter13 = project dependsOn (common)

chapter12/src/main/scala/com/reactivedesignpatterns/chapter12/ActiveActive.scala

@@ -32,6 +32,7 @@ object ActiveActive {
 
     def initialized: Receive = {
       case SeqCommand(seq, cmd, replyTo) =>
+        // tracking of sequence numbers and resends is elided here
         cmd match {
           case Put(key, value, r) =>
             map += key -> value
@@ -53,18 +54,19 @@ object ActiveActive {
     def add(res: SeqResult): ReplyState
     def isFinished: Boolean = missing.isEmpty
   }
-  private case class Unknown(deadline: Deadline, replies: Set[SeqResult], missing: Set[ActorRef]) extends ReplyState {
+  private case class Unknown(deadline: Deadline, replies: Set[SeqResult], missing: Set[ActorRef], quorum: Int) extends ReplyState {
     override def add(res: SeqResult): ReplyState = {
-      val quorum = (missing.size + 1) / 2
       val nextReplies = replies + res
+      val nextMissing = missing - res.replica
       if (nextReplies.size >= quorum) {
         val answer = replies.toSeq.groupBy(_.res).collectFirst { case (k, s) if s.size >= quorum => s.head }
         if (answer.isDefined) {
           val right = answer.get
           val wrong = replies.collect { case SeqResult(_, res, replica, _) if res != right => replica }
-          Known(deadline, right, wrong, missing - res.replica)
-        } else Unknown(deadline, nextReplies, missing - res.replica)
-      } else Unknown(deadline, nextReplies, missing - res.replica)
+          Known(deadline, right, wrong, nextMissing)
+        } else if (nextMissing.isEmpty) Known.fromUnknown(deadline, nextReplies)
+        else Unknown(deadline, nextReplies, nextMissing, quorum)
+      } else Unknown(deadline, nextReplies, nextMissing, quorum)
     }
   }
   private case class Known(deadline: Deadline, reply: SeqResult, wrong: Set[ActorRef], missing: Set[ActorRef]) extends ReplyState {
@@ -73,6 +75,18 @@ object ActiveActive {
       Known(deadline, reply, nextWrong, missing - res.replica)
     }
   }
+  private object Known {
+    def fromUnknown(deadline: Deadline, replies: Set[SeqResult]): Known = {
+      // did not reach consensus on this one, pick simple majority
+      val counts = replies.groupBy(_.res)
+      val biggest = counts.iterator.map(_._2.size).max
+      val winners = counts.collectFirst {
+        case (res, win) if win.size == biggest => win
+      }.get
+      val losers = (replies -- winners).map(_.replica)
+      Known(deadline, winners.head, losers, Set.empty)
+    }
+  }
 
   class Coordinator(N: Int) extends Actor {
     private var replicas = (1 to N).map(_ => newReplica()).toSet
@@ -81,7 +95,7 @@ object ActiveActive {
     private var nextReply = 0
 
     override def supervisorStrategy = SupervisorStrategy.stoppingStrategy
-    
+
     private def newReplica(): ActorRef =
       context.watch(context.actorOf(Replica.props))
 
@@ -92,7 +106,7 @@ object ActiveActive {
       case cmd: Command =>
         val c = SeqCommand(seqNr.next, cmd, self)
         replicas foreach (_ ! c)
-        replies += c.seq -> Unknown(5 seconds fromNow, Set.empty, replicas)
+        replies += c.seq -> Unknown(5 seconds fromNow, Set.empty, replicas, (replicas.size + 1) / 2)
       case res: SeqResult if replies.contains(res.seq) && replicas.contains(res.replica) =>
         val prevState = replies(res.seq)
         val nextState = prevState.add(res)
@@ -111,18 +125,10 @@ object ActiveActive {
       val expired = replies.iterator.takeWhile(_._2.deadline <= now)
       for ((seq, state) <- expired) {
         state match {
-          case Unknown(deadline, received, missing) =>
-            // did not reach consensus on this one, pick simple majority
-            val counts = received.groupBy(_.res)
-            val biggest = counts.iterator.map(_._2.size).max
-            val winners = counts.collectFirst {
-              case (res, win) if win.size == biggest => win
-            }.get
-            val losers = (received -- winners).map(_.replica)
-            // don’t wait for further responses
-            replies += seq -> Known(deadline, winners.head, losers, Set.empty)
+          case Unknown(deadline, received, _, _) =>
+            val forced = Known.fromUnknown(deadline, received)
+            replies += seq -> forced
           case Known(deadline, reply, wrong, missing) =>
-            // don’t wait for further responses
             replies += seq -> Known(deadline, reply, wrong, Set.empty)
         }
       }

chapter13/build.sbt

@@ -0,0 +1,3 @@
+import Build._
+
+libraryDependencies ++= Seq(akkaActor, amazonAWS, scalaJava8, akkaTestkit, junit, scalatest)

chapter13/src/main/scala/com/reactivedesignpatterns/chapter13/EC2Worker.java

@@ -0,0 +1,57 @@
+/**
+ * Copyright (C) 2015 Roland Kuhn <http://rolandkuhn.com>
+ */
+package com.reactivedesignpatterns.chapter13;
+
+import java.util.List;
+
+import scala.PartialFunction;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import akka.dispatch.Futures;
+import akka.japi.pf.PFBuilder;
+import akka.pattern.CircuitBreaker;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.services.ec2.AmazonEC2Client;
+import com.amazonaws.services.ec2.model.Instance;
+import com.amazonaws.services.ec2.model.Reservation;
+import com.amazonaws.services.ec2.model.RunInstancesRequest;
+import com.amazonaws.services.ec2.model.RunInstancesResult;
+
+public class EC2Worker {
+	public Instance startInstance(AWSCredentials credentials) {
+		AmazonEC2Client amazonEC2Client = new AmazonEC2Client(credentials);
+
+		RunInstancesRequest runInstancesRequest = new RunInstancesRequest()
+				.withImageId("").withInstanceType("m1.small").withMinCount(1)
+				.withMaxCount(1);
+
+		RunInstancesResult runInstancesResult = amazonEC2Client
+				.runInstances(runInstancesRequest);
+
+		Reservation reservation = runInstancesResult.getReservation();
+		List<Instance> instances = reservation.getInstances();
+
+		// there will be exactly one instance in this list, otherwise
+		// runInstances() would have thrown an exception
+		return instances.get(0);
+	}
+
+	private ExecutionContext executionContext;
+	private CircuitBreaker circuitBreaker;
+
+	public Future<Instance> startInstanceAsync(AWSCredentials credentials) {
+		Future<Instance> f = circuitBreaker.callWithCircuitBreaker(() ->
+			Futures.future(() -> startInstance(credentials), executionContext));
+		PartialFunction<Throwable, Future<Instance>> recovery =
+			new PFBuilder<Throwable, Future<Instance>>()
+				.match(AmazonClientException.class,
+					   ex -> ex.isRetryable(),
+					   ex -> startInstanceAsync(credentials))
+				.build();
+		return f.recoverWith(recovery, executionContext);
+	}
+
+}

project/Build.scala

@@ -8,9 +8,12 @@ object Build {
   val akkaContrib = "com.typesafe.akka" %% "akka-contrib" % akkaVersion
   val akkaDData = "com.typesafe.akka" %% "akka-distributed-data-experimental" % akkaVersion
 
+  val amazonAWS = "com.amazonaws" % "aws-java-sdk" % "1.10.14"
+
   val sbtIO = "org.scala-sbt" %% "io" % "0.13.8"
 
   val scalaAsync = "org.scala-lang.modules" %% "scala-async" % "0.9.2"
+  val scalaJava8 = "org.scala-lang.modules" %% "scala-java8-compat" % "0.6.0" exclude("org.scala-lang.modules", "fngen_2.11")
 
   val playJson = "com.typesafe.play" %% "play-json" % "2.4.0"