=doc #3639 Show complete custom mailbox example.

akka-docs/rst/scala/code/docs/dispatcher/DispatcherDocSpec.scala

@@ -13,6 +13,7 @@ import akka.event.LoggingAdapter
 import scala.concurrent.duration._
 import akka.actor._
 import docs.dispatcher.DispatcherDocSpec.MyBoundedActor
+import akka.dispatch.RequiresMessageQueue
 
 object DispatcherDocSpec {
   val javaConfig = """
@@ -29,6 +30,22 @@ object DispatcherDocSpec {
       //Other mailbox configuration goes here
     }
     //#prio-mailbox-config-java
+
+    //#custom-mailbox-config-java
+    custom-dispatcher {
+      mailbox-requirement =
+      "docs.dispatcher.MyUnboundedJMessageQueueSemantics"
+    }
+
+    akka.actor.mailbox.requirements {
+      "docs.dispatcher.MyUnboundedJMessageQueueSemantics" =
+      custom-dispatcher-mailbox
+    }
+
+    custom-dispatcher-mailbox {
+      mailbox-type = "docs.dispatcher.MyUnboundedJMailbox"
+    }
+    //#custom-mailbox-config-java
   """
 
   val config = """
@@ -154,6 +171,21 @@ object DispatcherDocSpec {
     }
     //#required-mailbox-config
 
+    //#custom-mailbox-config
+    custom-dispatcher {
+      mailbox-requirement =
+      "docs.dispatcher.MyUnboundedMessageQueueSemantics"
+    }
+
+    akka.actor.mailbox.requirements {
+      "docs.dispatcher.MyUnboundedMessageQueueSemantics" =
+      custom-dispatcher-mailbox
+    }
+
+    custom-dispatcher-mailbox {
+      mailbox-type = "docs.dispatcher.MyUnboundedMailbox"
+    }
+    //#custom-mailbox-config
   """
 
   //#prio-mailbox
@@ -195,33 +227,22 @@ object DispatcherDocSpec {
     with RequiresMessageQueue[BoundedMessageQueueSemantics]
   //#required-mailbox-class
 
-  //#mailbox-implementation-example
-  class MyUnboundedMailbox extends akka.dispatch.MailboxType {
-    import akka.actor.{ ActorRef, ActorSystem }
-    import com.typesafe.config.Config
-    import java.util.concurrent.ConcurrentLinkedQueue
-    import akka.dispatch.{
-      Envelope,
-      MessageQueue,
-      UnboundedQueueBasedMessageQueue
+  //#require-mailbox-on-actor
+  class MySpecialActor extends Actor
+    with RequiresMessageQueue[MyUnboundedMessageQueueSemantics] {
+    //#require-mailbox-on-actor
+    def receive = {
+      case _ ⇒
     }
-
-    // This constructor signature must exist, it will be called by Akka
-    def this(settings: ActorSystem.Settings, config: Config) = this()
-
-    // The create method is called to create the MessageQueue
-    final override def create(owner: Option[ActorRef],
-                              system: Option[ActorSystem]): MessageQueue =
-      new UnboundedQueueBasedMessageQueue {
-        final val queue = new ConcurrentLinkedQueue[Envelope]()
-      }
+    //#require-mailbox-on-actor
+    // ...
   }
-  //#mailbox-implementation-example
+  //#require-mailbox-on-actor
 }
 
 class DispatcherDocSpec extends AkkaSpec(DispatcherDocSpec.config) {
 
-  import DispatcherDocSpec.MyActor
+  import DispatcherDocSpec._
 
   "defining dispatcher in config" in {
     val context = system
@@ -325,4 +346,13 @@ class DispatcherDocSpec extends AkkaSpec(DispatcherDocSpec.config) {
     val dispatcher = system.dispatchers.lookup("my-balancing-dispatcher")
   }
 
+  "require custom mailbox on dispatcher" in {
+    val myActor = system.actorOf(Props[MyActor].withDispatcher(
+      "custom-dispatcher"))
+  }
+
+  "require custom mailbox on actor" in {
+    val myActor = system.actorOf(Props[MySpecialActor])
+  }
+
 }

akka-docs/rst/scala/code/docs/dispatcher/MyUnboundedMailbox.scala

@@ -0,0 +1,59 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.dispatcher
+
+//#mailbox-implementation-example
+import akka.actor.ActorRef
+import akka.actor.ActorSystem
+import akka.dispatch.Envelope
+import akka.dispatch.MailboxType
+import akka.dispatch.MessageQueue
+import akka.dispatch.ProducesMessageQueue
+import com.typesafe.config.Config
+import java.util.concurrent.ConcurrentLinkedQueue
+import scala.Option
+
+// Marker trait used for mailbox requirements mapping
+trait MyUnboundedMessageQueueSemantics
+
+object MyUnboundedMailbox {
+  // This is the MessageQueue implementation
+  class MyMessageQueue extends MessageQueue
+    with MyUnboundedMessageQueueSemantics {
+
+    private final val queue = new ConcurrentLinkedQueue[Envelope]()
+
+    // these must be implemented; queue used as example
+    def enqueue(receiver: ActorRef, handle: Envelope): Unit =
+      queue.offer(handle)
+    def dequeue(): Envelope = queue.poll()
+    def numberOfMessages: Int = queue.size
+    def hasMessages: Boolean = !queue.isEmpty
+    def cleanUp(owner: ActorRef, deadLetters: MessageQueue) {
+      while (hasMessages) {
+        deadLetters.enqueue(owner, dequeue())
+      }
+    }
+  }
+}
+
+// This is the Mailbox implementation
+class MyUnboundedMailbox extends MailboxType
+  with ProducesMessageQueue[MyUnboundedMailbox.MyMessageQueue] {
+
+  import MyUnboundedMailbox._
+
+  // This constructor signature must exist, it will be called by Akka
+  def this(settings: ActorSystem.Settings, config: Config) = {
+    // put your initialization code here
+    this()
+  }
+
+  // The create method is called to create the MessageQueue
+  final override def create(owner: Option[ActorRef],
+                            system: Option[ActorSystem]): MessageQueue =
+    new MyMessageQueue()
+}
+//#mailbox-implementation-example