In this recipe, you will learn how to create a custom mailbox for an actor. As you're already aware, in Akka, each actor has its own mailbox-like queue from which it picks up the messages one by one, and processes them. There are some custom mailbox implementations provided by Akka, such as PriorityMailbox and controlAwareMailbox, other than the default mailbox.
There might be a situation when you want to control the way the actor picks up the message or anything else. We will create an actor mailbox that will accept messages from actors of a particular name.
To step through this recipe, we need to import our Hello-Akka
project in the IDE-like intelliJ Idea. Prerequisites are the same as those in the previous recipes.
- Create a Scala file, say
CustomMailbox.scala
, in packagecom.packt.chapter1
.
Add the following required imports to the top of the file:
import java.util.concurrent.ConcurrentLinkedQueue import akka.actor.{Props, Actor, ActorSystem,ActorRef} import akka.dispatch.{ MailboxType, ProducesMessageQueue, Envelope, MessageQueue} import com.typesafe.config.Config
- Define a
MyMessageQueue
, which extends traitMessageQueue
and implementing methods:
class MyMessageQueue extends MessageQueue { private final val queue = new ConcurrentLinkedQueue[Envelope]() // these should be implemented; queue used as example def enqueue(receiver: ActorRef, handle: Envelope): Unit = { if(handle.sender.path.name == "MyActor") { handle.sender ! "Hey dude, How are you?, I Know your name,processing your request" queue.offer(handle) } else handle.sender ! "I don't talk to strangers, I can't process your request" } 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()) } } }
- Let's provide a custom mailbox implementation, which uses the preceding
MessageQueue
:
class MyUnboundedMailbox extends MailboxType with ProducesMessageQueue[MyMessageQueue] { 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 MyMessageQueue() }
- Create an
application.conf
file and put the below configuration. Anapplication.conf
file is used to configure Akka application properties and it resides in the project's resource directory.
custom-dispatcher { mailbox-requirement = "com.packt.chapter1.MyMessageQueue" } akka.actor.mailbox.requirements { "com.packt.chapter1.MyMessageQueue" = custom-dispatcher- mailbox } custom-dispatcher-mailbox { mailbox-type = "com.packt.chapter1.MyUnboundedMailbox" }
- Now define an actor that would use the preceding configuration, say,
MySpecialActor
. It's special, because it would talk to the actor whom it knows, and say hello to that actor only:
class MySpecialActor extends Actor { override def receive: Receive = { case msg: String => println(s"msg is $msg" ) } }
- Define an actor who will try to talk to the special actor:
class MyActor extends Actor { override def receive: Receive = { case (msg: String, actorRef: ActorRef) => actorRef ! msg case msg => println(msg) } }
- Create a test application,
CustomMailbox
, as follows:
object CustomMailbox extends App { val actorSystem = ActorSystem("HelloAkka") val actor = actorSystem.actorOf(Props[MySpecialActor].withDispatcher ("custom-dispatcher")) val actor1 = actorSystem.actorOf(Props[MyActor],"xyz") val actor2 = actorSystem.actorOf(Props[MyActor],"MyActor") actor1 ! ("hello", actor) actor2 ! ("hello", actor) }
- Run the application in the IDE or from the console, and you will get the following output:
I don't talk to strangers, I can't process your request Hey dude, How are you?, I Know your name,processing your request msg is hello
As you know, a mailbox uses a message queue, and we need to provide a custom implementation for the queue.
In step two, we define a class, MyMessageQueue
, which extends the trait MessageQueue
and the implementing methods.
We want our actor to receive messages from only those actors whose name is MyActor
, and not from any other actor.
To achieve the aforementioned functionality, we implement the enqueue
method, and specify that the message should be enqueued if sender name is MyActor
, otherwise ignore the message.
In this case, we used ConcurrentLinkedQueue
as the underlying data structure for the queue.
However, it is up to us which data structure we pick for enqueing and removing messages. Changing the data structure may also change the processing order of messages.
In step three, we define the custom mailbox using MyMessageQueue
.
In step four, we configure the preceding mailbox with a custom-dispatcher
in application.conf
.
In step five and six, we define MySpecialActor
, which will use the custom mailbox when we create it with the custom-dispatcher
. MyActor
is the actor which tries to communicate with MySpecialActor
.
In step seven, we have two instances of MyActor
, actor1
and actor2
, which send messages to MySpecialActor
.
Since MySpecialActor
talks to only those Actors whose name is MyActor
, it does not process messages from MyActor
whose name is xyz
, as you can see in the output.