update Scala/Akka and make compile, #3

Author: rkuhn

build.sbt

@@ -1,6 +1,6 @@
 organization in ThisBuild := "com.reactivedesignpatterns"
 version in ThisBuild := "0.1-SNAPSHOT"
-scalaVersion in ThisBuild := "2.11.6"
+scalaVersion in ThisBuild := "2.11.8"
 
 lazy val common = project
 

chapter13/src/main/scala/com/reactivedesignpatterns/chapter13/RateLimiter.scala

@@ -27,30 +27,3 @@ class RateLimiter(requests: Int, period: FiniteDuration) {
     }
   }
 }
-
-object CircuitBreaker {
-  private object StorageFailed extends RuntimeException
-  private def sendToStorage(job: Job): Future[StorageStatus] = {
-    // make an asynchronous request to the storage subsystem
-    val f: Future[StorageStatus] = ???
-    // map storage failures to Future failures to alert the breaker
-    f.map {
-      case StorageStatus.Failed => throw StorageFailed
-      case other                => other
-    }
-  }
-  private val breaker = CircuitBreaker(
-    system.scheduler, // used for scheduling timeouts
-    5, // number of failures in a row when it trips
-    300.millis, // timeout for each service call
-    30.seconds // time before trying to close after tripping
-    )
-  def persist(job: Job): Future[StorageStatus] =
-    breaker
-      .withCircuitBreaker(sendToStorage(job))
-      .recover {
-        case StorageFailed                  => StorageStatus.Failed
-        case _: TimeoutException            => StorageStatus.Unknown
-        case _: CircuitBreakerOpenException => StorageStatus.Failed
-      }
-}

chapter14/src/main/java/com/reactivedesignpatterns/chapter14/ComplexCommand.java

@@ -6,16 +6,18 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Set;
+import java.util.concurrent.CompletionStage;
 import java.util.stream.Stream;
 
 import javax.script.Invocable;
 import javax.script.ScriptEngine;
 import javax.script.ScriptEngineManager;
 
+import akka.NotUsed;
 import akka.japi.Pair;
 import akka.japi.function.*;
 import akka.stream.javadsl.*;
-import akka.stream.scaladsl.Keep;
+import akka.stream.javadsl.Keep;
 import scala.concurrent.Future;
 import scala.runtime.BoxedUnit;
 
@@ -151,10 +153,10 @@ public T apply(T arg0, DataElement arg1) throws Exception {
 
 	class Inject<T> implements Function<DataElement, DataElement> {
 		private static final long serialVersionUID = 1L;
-		public final RunnableGraph<Future<T>> value;
+		public final RunnableGraph<CompletionStage<T>> value;
 		public final String fieldname;
 
-		public Inject(RunnableGraph<Future<T>> value, String fieldname) {
+		public Inject(RunnableGraph<CompletionStage<T>> value, String fieldname) {
 			this.value = value;
 			this.fieldname = fieldname;
 		}
@@ -198,10 +200,10 @@ public Set<T> apply(Set<T> arg0, DataElement arg1) throws Exception {
 	}
 
 	public static void akkaStreamDSL() {
-		RunnableGraph<Future<Long>> p =
+		RunnableGraph<CompletionStage<Long>> p =
 			Source.<DataElement>empty()
 			.filter(new InRange("year", 1950, 1960))
-			.toMat(Sink.fold(0L, new Median<Long>("price")), Keep.<BoxedUnit, Long>right());
+			.toMat(Sink.fold(0L, new Median<Long>("price")), Keep.<NotUsed, CompletionStage<Long>>right());
 
 		Source.<DataElement>empty()
 			.map(new Inject<Long>(p, "p"))

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

@@ -9,6 +9,7 @@ import akka.typed.AskPattern._
 import scala.concurrent.duration._
 import akka.util.Timeout
 import akka.pattern.AskTimeoutException
+import akka.actor.ReceiveTimeout
 
 object Aggregator {
 
@@ -59,7 +60,7 @@ object Aggregator {
     ContextAware { ctx =>
       Static {
         case GetFrontPage(user, replyTo) =>
-          val childRef = ctx.spawnAnonymous(Props {
+          val childRef = ctx.spawnAnonymous(Deferred { () =>
             val builder = new FrontPageResultBuilder(user)
             Partial[AnyRef](
               pf {
@@ -85,6 +86,7 @@ object Aggregator {
     ContextAware { ctx =>
       import ctx.executionContext
       implicit val timeout = Timeout(1.second)
+      implicit val scheduler = ctx.system.scheduler
 
       Static {
         case GetFrontPage(user, replyTo) =>
@@ -123,6 +125,7 @@ object Aggregator {
     ContextAware { ctx =>
       import ctx.executionContext
       implicit val timeout = Timeout(1.second)
+      implicit val scheduler = ctx.system.scheduler
 
       Static {
         case GetFrontPage(user, replyTo) =>

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

@@ -27,8 +27,10 @@ object AskPattern {
                        body: String, correlationID: UUID,
                        replyTo: ActorRef[SendEmailResult])
 
+  sealed trait Result
+  case object ReceiveTimeout extends Result
   case class SendEmailResult(correlationID: UUID, status: StatusCode,
-                             explanation: String)
+                             explanation: String) extends Result
 
   sealed trait StatusCode
   object StatusCode {
@@ -38,7 +40,7 @@ object AskPattern {
 
   def withoutAskPattern(emailGateway: ActorRef[SendEmail]): Behavior[StartVerificationProcess] =
     ContextAware[MyCommands] { ctx =>
-      val log = Logging(ctx.system.eventStream, "VerificationProcessManager")
+      val log = new BusLogging(ctx.system.eventStream, "VerificationProcessManager", getClass, ctx.system.logFilter)
       var statusMap = Map.empty[UUID, (String, ActorRef[VerificationProcessResponse])]
       val adapter = ctx.spawnAdapter((s: SendEmailResult) => MyEmailResult(s.correlationID, s.status, s.explanation))
 
@@ -69,16 +71,16 @@ object AskPattern {
 
   def withChildActor(emailGateway: ActorRef[SendEmail]): Behavior[StartVerificationProcess] =
     ContextAware { ctx: ActorContext[StartVerificationProcess] =>
-      val log = Logging(ctx.system.eventStream, "VerificationProcessManager")
+      val log = new BusLogging(ctx.system.eventStream, "VerificationProcessManager", getClass, ctx.system.logFilter)
 
       Static {
         case StartVerificationProcess(userEmail, replyTo) =>
           val corrID = UUID.randomUUID()
-          val childActor = ctx.spawnAnonymous(Props(FullTotal[SendEmailResult] {
+          val childActor = ctx.spawnAnonymous(FullTotal[Result] {
             case Sig(ctx, PreStart) =>
-              ctx.setReceiveTimeout(5.seconds)
+              ctx.setReceiveTimeout(5.seconds, ReceiveTimeout)
               Same
-            case Sig(_, ReceiveTimeout) =>
+            case Msg(_, ReceiveTimeout) =>
               log.warning("verification process initiation timed out for {}", userEmail)
               replyTo ! VerificationProcessFailed(userEmail)
               Stopped
@@ -93,17 +95,18 @@ object AskPattern {
             case Msg(_, SendEmailResult(wrongID, _, _)) =>
               log.error("received wrong SendEmailResult for corrID {}", corrID)
               Same
-          }))
+          })
           val request = SendEmail("verification@example.com", List(userEmail), constructBody(userEmail, corrID), corrID, childActor)
           emailGateway ! request
       }
     }
 
   def withAskPattern(emailGateway: ActorRef[SendEmail]): Behavior[StartVerificationProcess] =
     ContextAware { ctx =>
-      val log = Logging(ctx.system.eventStream, "VerificationProcessManager")
+      val log = new BusLogging(ctx.system.eventStream, "VerificationProcessManager", getClass, ctx.system.logFilter)
       implicit val timeout = Timeout(5.seconds)
       import ctx.executionContext
+      implicit val scheduler = ctx.system.scheduler
 
       Static {
         case StartVerificationProcess(userEmail, replyTo) =>

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

@@ -142,7 +142,7 @@ object PersistentBusinessHandshake extends App {
       var deliveryId: Long = 0
       deliver(alice)(id => { deliveryId = id; ChangeBudget(-amount, self, persistenceId) })
 
-      LoggingReceive {
+      LoggingReceive({
         case ChangeBudgetDone =>
           persist(AliceConfirmedChange(deliveryId)) { ev =>
             confirmDelivery(ev.deliveryId)
@@ -153,7 +153,7 @@ object PersistentBusinessHandshake extends App {
             confirmDelivery(ev.deliveryId)
             context.stop(self)
           }
-      }
+      }: Receive)
     }
 
     def talkToBob() = {

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

@@ -30,12 +30,12 @@ object RequestResponseTypedActors {
     }
 
   def main(args: Array[String]): Unit = {
-    ActorSystem("ReqResTyped", Props(ContextAware[Unit] { ctx =>
-      val res = ctx.spawn(Props(responder), "responder")
-      val req = ctx.watch(ctx.spawn(Props(requester(res)), "requester"))
+    ActorSystem("ReqResTyped", ContextAware[Unit] { ctx =>
+      val res = ctx.spawn(responder, "responder")
+      val req = ctx.watch(ctx.spawn(requester(res), "requester"))
       Full {
         case Sig(ctx, Terminated(`req`)) => Stopped
       }
-    }))
+    })
   }
-}
+}

project/Build.scala

@@ -1,21 +1,21 @@
 import sbt._
 
 object Build {
-  val akkaVersion = "2.4.1"
+  val akkaVersion = "2.4.17"
 
   val akkaActor   = "com.typesafe.akka" %% "akka-actor" % akkaVersion
   val akkaTestkit = "com.typesafe.akka" %% "akka-testkit" % akkaVersion % "test"
   val akkaContrib = "com.typesafe.akka" %% "akka-contrib" % akkaVersion
   val akkaSharding= "com.typesafe.akka" %% "akka-cluster-sharding" % akkaVersion
   val akkaDData   = "com.typesafe.akka" %% "akka-distributed-data-experimental" % akkaVersion
-  val akkaStream  = "com.typesafe.akka" %% "akka-stream-experimental" % "2.0.1"
+  val akkaStream  = "com.typesafe.akka" %% "akka-stream" % akkaVersion
   val akkaTyped   = "com.typesafe.akka" %% "akka-typed-experimental" % akkaVersion
   val akkaPersistence = "com.typesafe.akka" %% "akka-persistence" % akkaVersion
   val akkaPersistenceQuery = "com.typesafe.akka" %% "akka-persistence-query-experimental" % akkaVersion
 
   val levelDb     = "org.iq80.leveldb" % "leveldb" % "0.7"
 
-  val amazonAWS = "com.amazonaws" % "aws-java-sdk" % "1.10.14"
+  val amazonAWS = "com.amazonaws" % "aws-java-sdk" % "1.11.109"
 
   val sbtIO = "org.scala-sbt" %% "io" % "0.13.8"