Akka + Akka TestKit + Specs2で子アクターの障害時ハンドリングをテストする。

環境

• Mac OS X Version 10.10.2
• Scala 2.11.5
• sbt 0.13.7

準備

/root/to/project/path
   |-- build.sbt
   |-- src
   |    |-- main
   |    |    |-- scala
   |    |    |    |-- FaultTolerance.scala
   |    |-- test
   |    |    |-- scala
   |    |    |    |-- FaultToleranceSpec.scala

build.sbt

name := "akka"                                                                                                                                                    

version := "1.0"

scalaVersion := "2.11.5"

scalacOptions ++= Seq("-Xlint", "-deprecation", "-unchecked", "-feature", "-Xelide-below", "ALL")

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-actor" % "2.3.9",
  "com.typesafe.akka" %% "akka-testkit" % "2.3.9",
  "org.specs2" %% "specs2" % "2.4.1"
)

実装

src/main/scala/FaultTolerance.scala

import akka.actor.{Actor, OneForOneStrategy, Props}
import akka.actor.SupervisorStrategy._

import scala.concurrent.duration._
import scala.language.postfixOps

class Supervisor extends Actor {
  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
      case _: ArithmeticException      => Resume
      case _: NullPointerException     => Restart
      case _: IllegalArgumentException => Stop
      case _: Exception                => Escalate
    }

  def receive = {
    case p: Props => sender ! context.actorOf(p)
  }
}

class Child extends Actor {

  var state = 0
  def receive = {
    case ex: Exception => throw ex
    case x: Int => state = x
    case "get" => sender ! state
  }
}

src/test/scala/FaultToleranceSpec.scala

import akka.actor.{Terminated, ActorRef, ActorSystem, Props}
import akka.testkit._

import org.specs2.mutable.{After, Specification}
import org.specs2.time.NoTimeConversions

import scala.concurrent.duration._

/* A tiny class that can be used as a Specs2 'context'. */
abstract class AkkaTestkitSpecs2Support extends TestKit(ActorSystem()) with After with ImplicitSender {
  // make sure we shut down the actor system after all tests have run
  def after = system.shutdown()
}

class FaultToleranceSpec extends Specification with NoTimeConversions {
  // forces all tests to be run sequentially
  sequential

  "FaultTolerance" should {
    "Resume & Restart" in new AkkaTestkitSpecs2Support {
      within(1 second) {
        val supervisor = system.actorOf(Props[Supervisor], "supervisor")
        supervisor ! Props[Child]
        val child = expectMsgType[ActorRef]
        child ! 42
        child ! new ArithmeticException
        child ! "get"
        val r1 = expectMsg(42)
        child ! new NullPointerException
        child ! "get"
        val r2 = expectMsg(0)
        (r1, r2) must_==((42, 0): (Int, Int))
      }
    }

    "Stop" in new AkkaTestkitSpecs2Support {
      within(1 second) {
        val supervisor = system.actorOf(Props[Supervisor], "supervisor")
        supervisor ! Props[Child]
        val child = expectMsgType[ActorRef]
        watch(child)
        child ! new IllegalArgumentException
        val r = expectMsgPF() { case Terminated(`child`) => "terminated" }
        r must_== "terminated"
      }
    }

    "Escalate" in new AkkaTestkitSpecs2Support {
      within(1 second) {
        val supervisor = system.actorOf(Props[Supervisor], "supervisor")
        supervisor ! Props[Child]
        val child = expectMsgType[ActorRef]
        watch(child)
        child ! "get"
        val r1 = expectMsg(0)
        child ! new Exception("CRASH")
        val r2 = expectMsgPF() {
          case t @ Terminated(`child`) if t.existenceConfirmed => "terminated"
        }
        (r1, r2) must_== ((0, "terminated"): (Int, String))
      }
    }
  }
}

実行

$ sbt '~test-only FaultToleranceSpec'

参考

http://doc.akka.io/docs/akka/2.3.9/scala/fault-tolerance.html
http://blog.xebia.com/2012/10/01/testing-akka-with-specs2/