Akka Cookbook

To step through this recipe, you will need SBT installed on your machine. No other prerequisites are required.

If you don't have SBT installed on your machine, please visit the SBT manual installation page(http://www.scala-sbt.org/release/docs/Manual-Installation.html), and follow the instructions for the operating system you have.

SBT is used for building Scala projects as we have Maven for Java projects. However, both SBT and Maven are capable of building Scala, Java projects.

In this book, we will build our projects on the Ubuntu Linux operating system.

Following screenshot shows the creation of the project directory:

Creating the project directory

      set name := "Hello-Akka"
      set version := "1.0"  
      set scalaVersion: ="2.11.7" 
      session save
      exit

This will create a build file called build.sbt and a target to put your class files into.

      libraryDependencies += "com.typesafe.akka" % 
        "akka-actor_2.11" % "2.4.4"

We can select a specific Akka actor version against a specific Scala version in the maven repository:

Adding Akka dependency to build file

The SBT (Simple Build tool), as the name suggests is a widely used tool for building Scala projects.

In step one, we create a project directory where we keep our source files, project build definition, and target, where class files are kept for runtime execution.

In step two, we create a project build definition file, called build.sbt, by executing simple sbt command.

In step three, we add a library dependency in the build file for the Akka actor to enable Akka capabilities.

In step four, we download the Akka dependency using the sbt update command, and our project is now ready for writing Akka-based applications.

This is how we can set up an Akka-based project.

If you are using IntelliJ Idea, then you must have Scala and the SBT plugin install on it.

It is very easy to create an ActorSystem in Akka:

        package com.packt.chapter1 
        Import akka.actor.ActorSystem 
        /** 
          * Created by user 
          */ 
        object HelloAkkaActorSystem extends App { 
          val actorSystem = ActorSystem("HelloAkka") 
          println(actorSystem) 
        } 

      akka://HelloAkka

Downloading the Akka actor dependency

In the recipe, we create a simple Scala object creating an ActorSystem, thereafter, we run the application.

To step through this recipe, we need to import the Hello-Akka project in any IDE, like intelliJ Idea, and ensure that SBT is installed on our machine to build and run the Scala project from the console.

Inside the file, define the actor as follows:

        class SummingActor extends Actor { 
          // state inside the actor 
          var sum = 0 
          // behaviour which is applied on the state 
          override def receive: Receive = { 
            // receives message an integer 
            case x: Int => sum = sum + x 
            println(s"my state as sum is $sum") 
            // receives default message 
            case _ => println("I don't know what
              are you talking about") 
          } 
        } 

Here, we are not creating an actor, we are only defining the state and behavior.

Add the following imports to the top of the file:

        import akka.actor.Props 
        import akka.actor.ActorSystem 
        object BehaviourAndState extends App {  
          val actorSystem = ActorSystem("HelloAkka") 
          // creating an actor inside the actor system 
          val actor = actorSystem.actorOf(Props[SummingActor]) 
          // print actor path  
          println(actor.path) 
        } 

You can run the application from the console using the following command:

      sbt "runMain com.packt.chapter1.BehaviorAndState"
      akka://HelloAkka/user/$a

Here, HelloAkka is the ActorSystem name, user is the user guardian actor, that is, the parent actor for your actor, and $a is the name of your actor.

You can also give a name to your actor:

        val actor = actorSystem.actorOf(Props[SummingActor],
        "summingactor") 

If you run the application again, it will print the actor name as summingactor.

The output will be as follows:

akka://HelloAkka/user/summingactor

How do we create an actor if it takes an argument in the constructor as shown in the following code:

        class SummingActorWithConstructor(intitalSum: Int)
        extends Actor { 
          // state inside the actor 
          var sum = 0 
          // behaviour which is applied on the state 
          override def receive: Receive = { 
            // receives message an integer 
            case x: Int => sum = intitalSum + sum + x 
            println(s"my state as sum is $sum") 
            // receives default message 
            case _ => println("I don't know what
              are you talking about") 
          } 
        } 

For this, we use the following code:

        actorSystem.actorOf(Props(classOf[
        SummingActorWithConstructor], 10), "summingactor")  

As we know from the previous recipe, the ActorSystem is a place where the actor lives.

In the preceding application, we define the actor with its state and behavior, and then create it inside Akka using the API provided by Akka.

In case of the summingactor, the state is the variable sum and the behavior is adding of the integer to the sum as soon as the message arrives.

There are some recommended practices related to creation of actors. For more details, you can check out the following link:

http://doc.akka.io/docs/akka/current/scala/actors.html.

In this recipe, we will learn how to send messages to actors. Prerequisites are the same as the previous recipes.

In the previous recipe, we created an actor which calculates the sum of integers:

        val actor = actorSystem.actorOf(Props[SummingActor],
          "summingactor") 

Now, we will send messages as integers to the summing actor, as follows:

        actor ! 1 

The following will be the output:

My state as sum is 1

If we keep on sending messages inside a while loop, the actor will continue to calculate the sum incrementally:

        while (true) { 
          Thread.sleep(3000) 
          actor ! 1 
        } 

On sending messages inside a while loop, the following output will be displayed:

my state as sum is 1
my state as sum is 2
my state as sum is 3
my state as sum is 4
my state as sum is 5

If we send a string message Hello, this message will fall into the actor's default behavior case, and the output will be as follows:

I don't know what you are talking about

Actors have methods to communicate with each other actors like tell (!) or ask (?) where the first one is fire and forget and the second returns a Future which means the response will come from that actor in the future.

As soon as you send the message to the actor, it receives the message, picks up an underlying Java thread from the thread pool, does it's work, and releases the thread. The actors never block your current thread of execution, thus, they are asynchronous by nature.

Visit the following link to see more information on send messages:

http://doc.akka.io/docs/akka/current/scala/actors.html#Send_messages.

Let's define an actor that computes something, say, the Fibonacci of a number:

        import akka.actor.Actor 

Now we define an actor which computes the Fibonacci of a number:

        class FibonacciActor extends Actor { 
          override def receive: Receive = { 
            case num : Int =>  
            val fibonacciNumber = fib(num) 
          } 
          def fib( n : Int) : Int = n match { 
            case 0 | 1 => n  
            case _ => fib( n-1 ) + fib( n-2 ) 
          } 
        } 

        sender ! fibonacciNumber 

Now, notice the difference:

        class FibonacciActor extends Actor { 
          override def receive: Receive = { 
            case num : Int => 
            val fibonacciNumber = fib(num) 
            sender ! fibonacciNumber 
          } 
          def fib( n : Int) : Int = n match { 
            case 0 | 1 => n 
            case _ => fib( n-1 ) + fib( n-2 ) 
          } 
        } 

Actors, by their implementation, know the default immediate sender, that is, they know who has sent them the message.

        import akka.actor.{Props, ActorSystem} 
        import akka.pattern.ask 
        import akka.util.Timeout 
        import scala.concurrent.Await 
        import scala.concurrent.duration._ 

        object FibonacciActorApp extends App {  
          implicit val timeout = Timeout(10 seconds) 
          val actorSystem = ActorSystem("HelloAkka") 
          val actor = actorSystem.actorOf(Props[FibonacciActor]) 
          // asking for result from actor 
          val future = (actor ? 10).mapTo[Int] 
          val fiboacciNumber = Await.result(future, 10 seconds) 
          println(fiboacciNumber) 
        } 

We create an actor that computes Fibonacci number, and sends the result to the sender who sent him the message to compute the Fibonacci.

In the actor receive block, we send the Fibonacci result back to the sender. Actors, by nature, know who has sent them the message, thus we always have the sender present in the context of the receive block.

When you send a message to the actor using a question mark (?), it returns a future promising that you will get the result when the operation would be completed.

We will learn about futures in later chapters.

To know more about sending messages to actors, go to the following link:

http://doc.akka.io/docs/akka/current/scala/actors.html#Send_messages.

To step through this recipe we will import the Hello-Akka project in our IDE and other prerequisites are same as before.

We will create the following two actors here:

The following are the steps for creating the actors:

        import akka.actor.ActorRef 

After adding the import add the code that follows:

        object Messages { 
          case class Done(randomNumber: Int) 
          case object GiveMeRandomNumber 
          case class Start(actorRef: ActorRef) 
        } 

        import akka.actor.Actor 
        import scala.util.Random._ 

Now add the code that follows:

        class RandomNumberGeneratorActor extends Actor { 
          import Messages._ 
          override def receive: Receive = { 
            case GiveMeRandomNumber => 
            println("received a message to 
              generate a random integer") 
            val randomNumber = nextInt 
            sender ! Done(randomNumber) 
          } 
        } 

        class QueryActor extends Actor { 
          import Messages._ 
          override def receive: Receive = { 
            case Start(actorRef) => println(s"send me the next 
             random number") 
            actorRef ! GiveMeRandomNumber 
            case Done(randomNumber) => 
            println(s"received a random number $randomNumber") 
          } 
        } 

        object Communication extends App { 
          import Messages._ 
          val actorSystem = ActorSystem("HelloAkka") 
          val randomNumberGenerator = 
           actorSystem.actorOf(Props[RandomNumberGeneratorActor], 
          "randomNumberGeneratorActor") 
          val queryActor = actorSystem.actorOf(Props[QueryActor], 
           "queryActor") 
          queryActor ! Start(randomNumberGenerator) 
        } 

      send me the next random number
      received a message to generate a random integer
      received a random number 841431704

In step two, we see there is message object, which contains messages to be sent to the actors. Actors will use these messages for communication.

In step three, we define RandomNumberGeneratorActor, which receives the message GiveMeRandomNumber, and sends it to the sender as follows:

    sender ! Done(randomNumber) 

In step four, we define QueryActor, which actually sends the message to RandomNumberGenerator, and receives the result in case of Done.

In step five, we create a test application to see the execution flow of the whole message.

In the following recipes, we will see how actors implement the master-slave work pulling pattern.

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.

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 

        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()) 
            } 
          } 
        } 

        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() 
        } 

        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" 
        } 

        class MySpecialActor extends Actor { 
          override def receive: Receive = { 
            case msg: String => println(s"msg is $msg" ) 
          } 
        } 

        class MyActor extends Actor { 
          override def receive: Receive = { 
            case (msg: String, actorRef: ActorRef) => actorRef ! 
             msg 
            case msg => println(msg) 
          } 
        } 

        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) 
        } 

      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.

To step through this recipe, we need to import our Hello-Akka project in IDE-like IntelliJ Idea. Prerequisites are the same as those in previous recipes.

        class MyPriorityActor extends Actor { 
          def receive: PartialFunction[Any, Unit] = { 
            // Int Messages 
            case x: Int => println(x) 
            // String Messages 
            case x: String => println(x) 
            // Long messages 
            case x: Long => println(x) 
            // other messages 
            case x => println(x) 
          } 
        } 

        class MyPriorityActorMailbox(settings:
          ActorSystem.Settings, config: Config) extends 
          UnboundedPriorityMailbox ( 
         // Create a new PriorityGenerator,
           lower prio means more important 
           PriorityGenerator { 
             // Int Messages 
             case x: Int => 1 
             // String Messages 
             case x: String => 0 
             // Long messages 
             case x: Long => 2 
             // other messages 
             case _ => 3 
        }) 

        prio-dispatcher {  
          mailbox-type = 
           "com.packt.chapter1..MyPriorityActorMailbox" 
        } 

        object PriorityMailBoxApp extends App { 
           val actorSystem = ActorSystem("HelloAkka") 
          val myPriorityActor = 
           actorSystem.actorOf(Props[MyPriorityActor].withDispatcher 
         ("prio-dispatcher")) 
          myPriorityActor ! 6.0 
          myPriorityActor ! 1 
          myPriorityActor ! 5.0 
          myPriorityActor ! 3 
          myPriorityActor ! "Hello" 
          myPriorityActor ! 5 
          myPriorityActor ! "I am priority actor" 
          myPriorityActor ! "I process string messages first,then 
           integer, long and others" 
        } 

      Hello
      I process string messages first,then integer, long and others
      I am priority actor
      1
      3
      5
      6.0
      5.0

In step two, we just define an actor which processes Int, Long, String, and other messages.

In step four, we configure a prio-dispatcher with this MyPriorityActorMailbox.

In step five, we create an actor which will use the prio-dispatcher.

In step six, as we can see in the output, the string messages are processed first, because they were given highest priority.

To step through this recipe, we need to import our Hello-Akka project in IDE-like IntelliJ Idea. Prerequisites are the same as those of the previous recipes.

        import akka.dispatch.ControlMessage 
        import akka.actor.{Props, Actor, ActorSystem} 

        case object MyControlMessage extends ControlMessage 

        class Logger extends Actor {  
          def receive = { 
            case MyControlMessage => println("Oh, I have to process
              Control message first") 
            case x => println(x.toString) 
          } 
        } 

         control-aware-dispatcher {  
          mailbox-type = 
           "akka.dispatch.UnboundedControlAwareMailbox" 
          //Other dispatcher configuration goes here 
        } 

        object ControlAwareMailbox extends App {  
          val actorSystem = ActorSystem("HelloAkka") 
          val actor = 
           actorSystem.actorOf(Props[Logger].withDispatcher( 
          "control-aware-dispatcher")) 
          actor ! "hello" 
          actor ! "how are" 
          actor ! "you?" 
          actor ! MyControlMessage 
        } 

      Oh, I have to process Control message first
      hello
      how are
      you?

In step three, we create an object, MyControlMessage, which extends the ControlMessage.

ControlMessage is a trait. The message which extends this trait will be handled on priority by ControlAwareMailbox. ControlAwareMailbox maintains two queues to allow messages that extend ControlMessage to be delivered with priority.

In step four, we create an actor which will handle ControlMessage.

In step five, we configure the control-aware-dispatcher in application.conf.

In step six. we create the actor with control-aware-dispatcher.

In step seven, we are able to see in the output that the actor processed ControlMessage first.

To step through this recipe, we need to import the Hello-Akka project in the IDE-like IntelliJ Idea. Prerequisites are the same as those in previous recipes.

        import akka.actor.{Props, ActorSystem, Actor} 

        class BecomeUnBecomeActor extends Actor { 
          def receive: Receive = { 
            case true => context.become(isStateTrue) 
            case false => context.become(isStateFalse) 
            case _ => println("don't know what you want to say !! ") 
          } 
          def isStateTrue: Receive  = { 
            case msg : String => println(s"$msg") 
            case false => context.become(isStateFalse) 
          } 
          def isStateFalse: Receive  = { 
            case msg : Int => println(s"$msg") 
            case true =>  context.become(isStateTrue) 
          } 
        } 

        object BecomeUnBecomeApp extends App { 
          val actorSystem = ActorSystem("HelloAkka") 
          val becomeUnBecome = 
           actorSystem.actorOf(Props[BecomeUnBecomeActor]) 
          becomeUnBecome ! true 
          becomeUnBecome ! "Hello how are you?" 
          becomeUnBecome ! false 
          becomeUnBecome ! 1100 
          becomeUnBecome ! true 
          becomeUnBecome ! "What do u do?" 
        } 

      Hello how are you?
      1100
      What do u do?

In step two, we define an actor, which changes its state to handle string and integer values.

If the state is true, we set the behavior as context.become(isStateTrue), and it starts handling string messages. If the state is false, we set the behavior as context.become(isStateFalse), and it starts handling integer messages.

In step four, we create the actor and send it to see if the output matches the functionality.

To step through this recipe, we need import the Hello-Akka project in an IDE like IntelliJ Idea. Prerequisites are the same as those in previous recipes.

        import akka.actor.{PoisonPill, Props, ActorSystem, Actor} 

        case object Stop 

        class ShutdownActor extends Actor { 
          override def receive: Receive = { 
            case msg:String => println(s"$msg") 
            case Stop => context.stop(self) 
          } 
        } 

Create an actor and send it a message as shown in the following code:

        object ShutdownApp extends App{ 
          val actorSystem = ActorSystem("HelloAkka") 
          val shutdownActor1 = 
           actorSystem.actorOf(Props[ShutdownActor], 
           "shutdownActor1") 
          shutdownActor1 ! "hello" 
          shutdownActor1 ! PoisonPill 
          shutdownActor1 ! "Are you there?" 
          val shutdownActor2 = 
           actorSystem.actorOf(Props[ShutdownActor], 
           "shutdownActor2") 
          shutdownActor2 ! "hello" 
          shutdownActor2 ! Stop 
          shutdownActor2 ! "Are you there?" 
        } 

      hello
      hello
      [INFO] [05/22/2016 20:39:53.137] [HelloAkka-akka.actor.default-
      dispatcher-4] [akka://HelloAkka/user/shutdownActor1] Message
      [java.lang.String] from Actor[akka://HelloAkka/deadLetters] to
      Actor[akka://HelloAkka/user/shutdownActor1#417818231] was not
      delivered. [1] dead letters encountered.
      [INFO] [05/22/2016 20:39:53.138] [HelloAkka-
      akka.actor.default-dispatcher-4]
      [akka://HelloAkka/user/shutdownActor2] Message
      [java.lang.String] from Actor[akka://HelloAkka/deadLetters]
      to Actor[akka://HelloAkka/user/shutdownActor2#788021817] was
      not delivered. [2] dead letters encountered.

In step three, we create a Stop message. Upon receiving this message, the actor will stop using context.stop(self).

In step four, we define an actor which handles the Stop message.

In step five, we create two actors of the same class, shutdownActor1 and shutdownActor2. We shut down shutdownActor1 using PoisonPill and shutdownActor2 using context.stop(self).

PoisonPill and context.stop(self) are the two ways to kill an actor. PoisonPill is the inbuilt message that is handled after all the messages that were already queued in the mailbox.

Context.stop is basically used for an ordered shutdown of actors when you want the child actors to stop first, then the parent actor, followed by the ActorSystem to stop top-level actors.