Merge branch 'master' of https://github.com/ReactiveDesignPatterns/CodeSamples

Author: jamie-allen

chapter04/src/main/scala/com/reactivedesignpatterns/chapter4/TranslationService.scala

@@ -2,12 +2,85 @@ package com.reactivedesignpatterns.chapter4
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
+import scala.concurrent.duration._
+import akka.actor.ActorRef
+import akka.actor.Actor
+import akka.util.Timeout
+import com.reactivedesignpatterns.Defaults._
+import java.util.concurrent.TimeoutException
+import akka.actor.Props
 
 class TranslationService {
   import ExecutionContext.Implicits.global
-  
+
   def translate(input: String): Future[String] = Future { Thread.sleep(100); "How are you?" }
-  
+
   def translate(input: String, ec: ExecutionContext): Future[String] =
     Future { Thread.sleep(100); "How are you?" }(ec)
 }
+
+/**
+ * Another implementation that is based on Actors; this is used in the example
+ * which shows how to assert the absence of indirectly invoked messages by way
+ * of a protocol adapter test for this translation service.
+ */
+object TranslationService {
+
+  /**
+   * Simplistic version 1 of the protocol: the reply will just be a String.
+   */
+  case class TranslateV1(query: String, replyTo: ActorRef)
+
+  /**
+   * Implementation of the TranslateV1 protocol.
+   */
+  private class TranslatorV1 extends Actor {
+    def receive = {
+      case TranslateV1(query, replyTo) =>
+        if (query == "sv:en:Hur mår du?") {
+          replyTo ! "How are you?"
+        } else {
+          replyTo ! s"error:cannot translate '$query'"
+        }
+    }
+  }
+  
+  def propsV1: Props = Props(new TranslatorV1)
+
+  /**
+   * More advanced version 2 of the protocol with proper reply types.
+   * Languages are communicated as Strings for brevity, in a real project
+   * these would be modeled as a proper Language type (statically known
+   * enumeration or based on runtime registration of values).
+   */
+  case class TranslateV2(phrase: String, inputLanguage: String, outputLanguage: String, replyTo: ActorRef)
+
+  sealed trait TranslationResponseV2
+  case class TranslationV2(inputPhrase: String, outputPhrase: String, inputLanguage: String, outputLanguage: String)
+  case class TranslationErrorV2(inputPhrase: String, inputLanguage: String, outputLanguage: String, errorMessage: String)
+
+  /**
+   * Implementation of the TranslateV2 protocol based on TranslatorV1.
+   */
+  private class TranslatorV2(v1: ActorRef) extends Actor {
+    implicit val timeout = Timeout(5.seconds)
+    import context.dispatcher
+
+    def receive = {
+      case TranslateV2(phrase, in, out, replyTo) =>
+        v1 ? (TranslateV1(s"$in:$out:$phrase", _)) collect {
+          case str: String =>
+            if (str.startsWith("error:")) {
+              TranslationErrorV2(phrase, in, out, str.substring(6))
+            } else {
+              TranslationV2(phrase, str, in, out)
+            }
+        } recover {
+          case _: TimeoutException =>
+            TranslationErrorV2(phrase, in, out, "timeout while talking to V1 back-end")
+        } pipeTo replyTo
+    }
+  }
+  
+  def propsV2(v1: ActorRef): Props = Props(new TranslatorV2(v1))
+}

chapter04/src/test/scala/com/reactivedesignpatterns/chapter4/EchoServiceSpec.scala

@@ -3,7 +3,6 @@ package com.reactivedesignpatterns.chapter4
 import scala.concurrent.duration.{ DurationInt, FiniteDuration }
 import scala.util.Try
 import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpec }
-import com.reactivedesignpatterns.Scoped.scoped
 import com.reactivedesignpatterns.Defaults._
 import akka.actor.{ Actor, ActorRef, ActorSystem, Props, actorRef2Scala }
 import akka.testkit.TestProbe
@@ -39,16 +38,13 @@ object EchoServiceSpec {
                 timings: List[FiniteDuration]): Receive = {
       case TimedResponse(Response(r), d) =>
         val start = outstanding(r)
-        scoped(
-          outstanding - r + sendRequest(echo, receiver),
-          (d - start) :: timings,
-          remaining - 1
-        ) { (outstanding, timings, remaining) =>
-            if (remaining > 0)
-              context.become(running(reportTo, echo, remaining, receiver, outstanding, timings))
-            else
-              context.become(finishing(reportTo, outstanding, timings))
-          }
+        val newOutstanding = outstanding - r + sendRequest(echo, receiver)
+        val newTimings = (d - start) :: timings
+        val newRemaining = remaining - 1
+        if (newRemaining > 0)
+          context.become(running(reportTo, echo, newRemaining, receiver, newOutstanding, newTimings))
+        else
+          context.become(finishing(reportTo, newOutstanding, newTimings))
       case AbortSLATest =>
         context.stop(self)
         reportTo ! SLAResponse(timings, outstanding)
@@ -57,12 +53,12 @@ object EchoServiceSpec {
     def finishing(reportTo: ActorRef, outstanding: Map[String, Timestamp], timings: List[FiniteDuration]): Receive = {
       case TimedResponse(Response(r), d) =>
         val start = outstanding(r)
-        scoped(outstanding - r, (d - start) :: timings) { (outstanding, timings) =>
-          if (outstanding.isEmpty) {
-            context.stop(self)
-            reportTo ! SLAResponse(timings, outstanding)
-          } else context.become(finishing(reportTo, outstanding, timings))
-        }
+        val newOutstanding = outstanding - r
+        val newTimings = (d - start) :: timings
+        if (newOutstanding.isEmpty) {
+          context.stop(self)
+          reportTo ! SLAResponse(newTimings, newOutstanding)
+        } else context.become(finishing(reportTo, newOutstanding, newTimings))
       case AbortSLATest =>
         context.stop(self)
         reportTo ! SLAResponse(timings, outstanding)
@@ -191,9 +187,9 @@ akka.actor.default-dispatcher.fork-join-executor.parallelism-max = 3
     }
 
   }
-  
+
   "An EchoService (with LatencyTestSupport)" should {
-    
+
     "keep its SLA" in {
       implicit val timeout = Timeout(5.seconds)
       import system.dispatcher
@@ -208,7 +204,7 @@ akka.actor.default-dispatcher.fork-join-executor.parallelism-max = 3
       lat.failureCount should be(0)
       lat.quantile(0.99) should be < 10.milliseconds
     }
-    
+
   }
 
 }

chapter04/src/test/scala/com/reactivedesignpatterns/chapter4/TranslationServiceSpec.scala

@@ -8,8 +8,51 @@ import scala.concurrent.ExecutionContext
 import scala.concurrent.Await
 import scala.concurrent.duration._
 import org.scalatest.concurrent.Eventually
+import org.scalatest.BeforeAndAfterAll
+import akka.actor.ActorSystem
+import akka.testkit.TestProbe
+import akka.actor.Actor
+import akka.actor.ActorRef
+import akka.actor.Props
 
-class TranslationServiceSpec extends WordSpec with Matchers with Eventually {
+object TranslationServiceSpec {
+  import TranslationService._
+
+  case object ExpectNominal
+  case object ExpectError
+  case class Unexpected(msg: Any)
+
+  class MockV1(reporter: ActorRef) extends Actor {
+    def receive = initial
+
+    override def unhandled(msg: Any) = {
+      reporter ! Unexpected(msg)
+    }
+
+    val initial: Receive = {
+      case ExpectNominal => context.become(expectingNominal)
+      case ExpectError   => context.become(expectingError)
+    }
+
+    val expectingNominal: Receive = {
+      case TranslateV1("sv:en:Hur mår du?", replyTo) =>
+        replyTo ! "How are you?"
+        context.become(initial)
+    }
+    
+    val expectingError: Receive = {
+      case TranslateV1(other, replyTo) =>
+        replyTo ! s"error:cannot parse input '$other'"
+        context.become(initial)
+    }
+  }
+
+  def mockV1props(reporter: ActorRef): Props = Props(new MockV1(reporter))
+
+}
+
+class TranslationServiceSpec extends WordSpec with Matchers with Eventually with BeforeAndAfterAll {
+  import TranslationServiceSpec._
 
   "A TranslationService" should {
 
@@ -22,15 +65,15 @@ class TranslationServiceSpec extends WordSpec with Matchers with Eventually {
         val ec = SynchronousEventLoop
         tr.translate(input, ec).value.get should be(Success(output))
       }
-      
+
       "using Await.result" in {
         val tr = new TranslationService
         val input = "Hur mår du?"
         val output = "How are you?"
         val future = tr.translate(input)
         Await.result(future, 1.second) should be(output)
       }
-      
+
       "using eventually()" in {
         val tr = new TranslationService
         val input = "Hur mår du?"
@@ -55,4 +98,66 @@ class TranslationServiceSpec extends WordSpec with Matchers with Eventually {
 
   }
 
+  implicit val system = ActorSystem("TranslationServiceSpec")
+  override def afterAll(): Unit = system.shutdown()
+
+  "A TranslationService version adapter" should {
+    import TranslationService._
+
+    "translate requests" in {
+      val v1 = TestProbe()
+      val v2 = system.actorOf(propsV2(v1.ref))
+      val client = TestProbe()
+
+      // initiate a request to the adapter
+      v2 ! TranslateV2("Hur mår du?", "sv", "en", client.ref)
+
+      // verify that the adapter asks the V1 service back-end
+      val req1 = v1.expectMsgType[TranslateV1]
+      req1.query should be("sv:en:Hur mår du?")
+
+      // initiate a reply
+      req1.replyTo ! "How are you?"
+
+      // verify that the adapter transforms it correctly
+      client.expectMsg(TranslationV2("Hur mår du?", "How are you?", "sv", "en"))
+
+      // now verify translation errors
+      v2 ! TranslateV2("Hur är läget?", "sv", "en", client.ref)
+      val req2 = v1.expectMsgType[TranslateV1]
+      // this implicitly verifies that no other communication happened with V1
+      req2.query should be("sv:en:Hur är läget?")
+      req2.replyTo ! "error:cannot parse input 'sv:en:Hur är läget?'"
+      client.expectMsg(TranslationErrorV2("Hur är läget?", "sv", "en", "cannot parse input 'sv:en:Hur är läget?'"))
+
+      v1.expectNoMsg(1.second)
+    }
+
+    "translate requests with async error reporting" in {
+      val asyncErrors = TestProbe()
+      val v1 = system.actorOf(mockV1props(asyncErrors.ref))
+      val v2 = system.actorOf(propsV2(v1))
+      val client = TestProbe()
+
+      // initiate a request to the adapter
+      v1 ! ExpectNominal
+      v2 ! TranslateV2("Hur mår du?", "sv", "en", client.ref)
+
+      // verify that the adapter transforms it correctly
+      client.expectMsg(TranslationV2("Hur mår du?", "How are you?", "sv", "en"))
+
+      // non-blocking check for async errors
+      asyncErrors.expectNoMsg(0.seconds)
+
+      // now verify translation errors
+      v1 ! ExpectError
+      v2 ! TranslateV2("Hur är läget?", "sv", "en", client.ref)
+      client.expectMsg(TranslationErrorV2("Hur är läget?", "sv", "en", "cannot parse input 'sv:en:Hur är läget?'"))
+
+      // final check for async errors
+      asyncErrors.expectNoMsg(1.second)
+    }
+
+  }
+
 }

common/src/main/scala/com/reactivedesignpatterns/Defaults.scala

@@ -4,16 +4,22 @@ import scala.concurrent.duration._
 import akka.actor.ActorRef
 import scala.concurrent.Future
 import akka.util.Timeout
+import scala.util.Success
 
 object Defaults {
-  
+
   val Timestamp = Deadline
   type Timestamp = Deadline
-  
+
   val SynchronousEventLoop = scala.concurrent.rdpextras.SynchronousEventLoop
 
   implicit class AskableActorRef(val ref: ActorRef) extends AnyVal {
     def ?(f: ActorRef => Any)(implicit timeout: Timeout): Future[Any] = akka.rdpextras.AskPattern.ask(ref, timeout, f)
   }
-  
+
+  implicit class PipeTo[T](val f: Future[T]) extends AnyVal {
+    def pipeTo(ref: ActorRef)(implicit self: ActorRef = ActorRef.noSender): Future[T] =
+      f.andThen { case Success(msg) => ref ! msg }(SynchronousEventLoop)
+  }
+
 }