Merging in the hotswap docs into master

akka-docs/scala/code/ActorDocSpec.scala

@@ -1,16 +1,19 @@
-package akka.docs.stm
+package akka.docs.actor
+
+//#imports1
+import akka.actor.Actor
+import akka.event.Logging
+//#imports1
+
+//#imports2
+import akka.actor.ActorSystem
+//#imports2
 
 import org.scalatest.{ BeforeAndAfterAll, WordSpec }
 import org.scalatest.matchers.MustMatchers
 import akka.testkit._
 import akka.util.duration._
 
-//#imports
-import akka.actor.Actor
-import akka.event.Logging
-
-//#imports
-
 //#my-actor
 class MyActor extends Actor {
   val log = Logging(context.system, this)
@@ -21,12 +24,110 @@ class MyActor extends Actor {
 }
 //#my-actor
 
+case class DoIt(msg: Message)
+case class Message(s: String)
+
+//#context-actorOf
+class FirstActor extends Actor {
+  val myActor = context.actorOf[MyActor]
+  //#context-actorOf
+  //#anonymous-actor
+  def receive = {
+    case m: DoIt ⇒
+      context.actorOf(new Actor {
+        def receive = {
+          case DoIt(msg) ⇒
+            val replyMsg = doSomeDangerousWork(msg)
+            sender ! replyMsg
+            self.stop()
+        }
+        def doSomeDangerousWork(msg: Message): String = { "done" }
+      }) ! m
+
+    case replyMsg: String ⇒ sender ! replyMsg
+  }
+  //#anonymous-actor
+}
+
+//#system-actorOf
+object Main extends App {
+  val system = ActorSystem("MySystem")
+  val myActor = system.actorOf[FirstActor]
+  //#system-actorOf
+}
+//#swapper
+case object Swap
+class Swapper extends Actor {
+  import context._
+  val log = Logging(system, this)
+
+  def receive = {
+    case Swap ⇒
+      log.info("Hi")
+      become {
+        case Swap ⇒
+          log.info("Ho")
+          unbecome() // resets the latest 'become' (just for fun)
+      }
+  }
+}
+
+object SwapperApp extends App {
+  val system = ActorSystem("SwapperSystem")
+  val swap = system.actorOf[Swapper]
+  swap ! Swap // logs Hi
+  swap ! Swap // logs Ho
+  swap ! Swap // logs Hi
+  swap ! Swap // logs Ho
+  swap ! Swap // logs Hi
+  swap ! Swap // logs Ho
+}
+//#swapper
+
+//#receive-orElse
+import akka.actor.Actor.Receive
+
+abstract class GenericActor extends Actor {
+  // to be defined in subclassing actor
+  def specificMessageHandler: Receive
+
+  // generic message handler
+  def genericMessageHandler: Receive = {
+    case event ⇒ printf("generic: %s\n", event)
+  }
+
+  def receive = specificMessageHandler orElse genericMessageHandler
+}
+
+class SpecificActor extends GenericActor {
+  def specificMessageHandler = {
+    case event: MyMsg ⇒ printf("specific: %s\n", event.subject)
+  }
+}
+
+case class MyMsg(subject: String)
+//#receive-orElse
+
 class ActorDocSpec extends AkkaSpec(Map("akka.loglevel" -> "INFO")) {
 
+  "import context" in {
+    //#import-context
+    class FirstActor extends Actor {
+      import context._
+      val myActor = actorOf[MyActor]
+      def receive = {
+        case x ⇒ myActor ! x
+      }
+    }
+    //#import-context
+
+    val first = system.actorOf(new FirstActor)
+    first.stop()
+
+  }
+
   "creating actor with AkkaSpec.actorOf" in {
-    //#creating-actorOf
     val myActor = system.actorOf[MyActor]
-    //#creating-actorOf
 
     // testing the actor
 
@@ -62,4 +163,73 @@ class ActorDocSpec extends AkkaSpec(Map("akka.loglevel" -> "INFO")) {
 
     myActor.stop()
   }
+
+  "creating actor with Props" in {
+    //#creating-props
+    import akka.actor.Props
+    val dispatcher = system.dispatcherFactory.fromConfig("my-dispatcher")
+    val myActor = system.actorOf(Props[MyActor].withDispatcher(dispatcher), name = "myactor")
+    //#creating-props
+
+    myActor.stop()
+  }
+
+  "using ask" in {
+    //#using-ask
+    class MyActor extends Actor {
+      def receive = {
+        case x: String ⇒ sender ! x.toUpperCase
+        case n: Int    ⇒ sender ! (n + 1)
+      }
+    }
+
+    val myActor = system.actorOf(new MyActor)
+    implicit val timeout = system.settings.ActorTimeout
+    val future = myActor ? "hello"
+    future.as[String] match {
+      case Some(answer) ⇒ //...
+      case None         ⇒ //...
+    }
+    val result: Option[Int] = for (x ← (myActor ? 3).as[Int]) yield { 2 * x }
+    //#using-ask
+
+    myActor.stop()
+  }
+
+  "using receiveTimeout" in {
+    //#receive-timeout
+    import akka.actor.ReceiveTimeout
+    import akka.util.duration._
+    class MyActor extends Actor {
+      context.receiveTimeout = Some(30 seconds)
+      def receive = {
+        case "Hello"        ⇒ //...
+        case ReceiveTimeout ⇒ throw new RuntimeException("received timeout")
+      }
+    }
+    //#receive-timeout
+  }
+
+  "using hot-swap" in {
+    //#hot-swap-actor
+    class HotSwapActor extends Actor {
+      import context._
+      def angry: Receive = {
+        case "foo" ⇒ sender ! "I am already angry?"
+        case "bar" ⇒ become(happy)
+      }
+
+      def happy: Receive = {
+        case "bar" ⇒ sender ! "I am already happy :-)"
+        case "foo" ⇒ become(angry)
+      }
+
+      def receive = {
+        case "foo" ⇒ become(angry)
+        case "bar" ⇒ become(happy)
+      }
+    }
+    //#hot-swap-actor
+  }
+
 }

akka-docs/scala/code/UnnestedReceives.scala

@@ -0,0 +1,48 @@
+package akka.docs.actor
+
+import akka.actor._
+import akka.actor.Actor._
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Requirements are as follows:
+ * The first thing the actor needs to do, is to subscribe to a channel of events,
+ * Then it must replay (process) all "old" events
+ * Then it has to wait for a GoAhead signal to begin processing the new events
+ * It mustn't "miss" events that happen between catching up with the old events and getting the GoAhead signal
+ */
+class UnnestedReceives extends Actor {
+  import context.become
+  //If you need to store sender/senderFuture you can change it to ListBuffer[(Any, Channel)]
+  val queue = new ListBuffer[Any]()
+
+  //This message processes a message/event
+  def process(msg: Any): Unit = println("processing: " + msg)
+  //This method subscribes the actor to the event bus
+  def subscribe() {} //Your external stuff
+  //This method retrieves all prior messages/events
+  def allOldMessages() = List()
+
+  override def preStart {
+    //We override preStart to be sure that the first message the actor gets is
+    //'Replay, that message will start to be processed _after_ the actor is started
+    self ! 'Replay
+    //Then we subscribe to the stream of messages/events
+    subscribe()
+  }
+
+  def receive = {
+    case 'Replay ⇒ //Our first message should be a 'Replay message, all others are invalid
+      allOldMessages() foreach process //Process all old messages/events
+      become { //Switch behavior to look for the GoAhead signal
+        case 'GoAhead ⇒ //When we get the GoAhead signal we process all our buffered messages/events
+          queue foreach process
+          queue.clear
+          become { //Then we change behaviour to process incoming messages/events as they arrive
+            case msg ⇒ process(msg)
+          }
+        case msg ⇒ //While we haven't gotten the GoAhead signal, buffer all incoming messages
+          queue += msg //Here you have full control, you can handle overflow etc
+      }
+  }
+}