Akka/Scala example source code file (PeekMailboxSpec.scala)

This example Akka source code file (PeekMailboxSpec.scala) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

actor, actorsystem, akka, check, die, dontwanna, doubleack, eventfilter, mail, peekactor, peekmailboxspec, poisonpill, test, testing, testkit

The PeekMailboxSpec.scala Akka example source code

/**
 * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
 */
package akka.contrib.mailbox

import com.typesafe.config.ConfigFactory

import akka.actor.{ Actor, ActorSystem, DeadLetter, PoisonPill, Props, Terminated, actorRef2Scala }
import akka.testkit.{ AkkaSpec, EventFilter, ImplicitSender }

object PeekMailboxSpec {
  case object Check
  case object DoubleAck
  class PeekActor(tries: Int) extends Actor {
    var togo = tries
    def receive = {
      case Check ⇒
        sender() ! Check
        PeekMailboxExtension.ack()
      case DoubleAck ⇒
        PeekMailboxExtension.ack()
        PeekMailboxExtension.ack()
      case msg ⇒
        sender() ! msg
        if (togo == 0) throw new RuntimeException("DONTWANNA")
        togo -= 1
        PeekMailboxExtension.ack()
    }
    override def preRestart(cause: Throwable, msg: Option[Any]) {
      for (m ← msg if m == "DIE") context stop self // for testing the case of mailbox.cleanUp
    }
  }
}

class PeekMailboxSpec extends AkkaSpec("""
    peek-dispatcher {
      mailbox-type = "akka.contrib.mailbox.PeekMailboxType"
      max-retries = 2
    }
    """) with ImplicitSender {

  import PeekMailboxSpec._

  "A PeekMailbox" must {

    "retry messages" in {
      val a = system.actorOf(Props(classOf[PeekActor], 1).withDispatcher("peek-dispatcher"))
      a ! "hello"
      expectMsg("hello")
      EventFilter[RuntimeException]("DONTWANNA", occurrences = 1) intercept {
        a ! "world"
      }
      expectMsg("world")
      expectMsg("world")
      a ! Check
      expectMsg(Check)
    }

    "put a bound on retries" in {
      val a = system.actorOf(Props(classOf[PeekActor], 0).withDispatcher("peek-dispatcher"))
      EventFilter[RuntimeException]("DONTWANNA", occurrences = 3) intercept {
        a ! "hello"
      }
      a ! Check
      expectMsg("hello")
      expectMsg("hello")
      expectMsg("hello")
      expectMsg(Check)
    }

    "not waste messages on double-ack()" in {
      val a = system.actorOf(Props(classOf[PeekActor], 0).withDispatcher("peek-dispatcher"))
      a ! DoubleAck
      a ! Check
      expectMsg(Check)
    }

    "support cleanup" in {
      system.eventStream.subscribe(testActor, classOf[DeadLetter])
      val a = system.actorOf(Props(classOf[PeekActor], 0).withDispatcher("peek-dispatcher"))
      watch(a)
      EventFilter[RuntimeException]("DONTWANNA", occurrences = 1) intercept {
        a ! "DIE" // stays in the mailbox
      }
      expectMsg("DIE")
      expectMsgType[DeadLetter].message should be("DIE")
      expectTerminated(a)
    }

  }

}

//#demo
class MyActor extends Actor {
  def receive = {
    case msg ⇒
      println(msg)
      doStuff(msg) // may fail
      PeekMailboxExtension.ack()
  }

  //#business-logic-elided
  var i = 0
  def doStuff(m: Any) {
    if (i == 1) throw new Exception("DONTWANNA")
    i += 1
  }

  override def postStop() {
    context.system.shutdown()
  }
  //#business-logic-elided
}

object MyApp extends App {
  val system = ActorSystem("MySystem", ConfigFactory.parseString("""
    peek-dispatcher {
      mailbox-type = "akka.contrib.mailbox.PeekMailboxType"
      max-tries = 2
    }
    """))

  val myActor = system.actorOf(Props[MyActor].withDispatcher("peek-dispatcher"),
    name = "myActor")

  myActor ! "Hello"
  myActor ! "World"
  myActor ! PoisonPill
}
//#demo