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

This example Akka source code file (ConsistentHashingRouterSpec.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

a, aa, actor, akka, any, b, concurrent, consistenthashableenvelope, consistenthashingrouterspec, msg, msg2, msgkey, string, test, testing, testkit

The ConsistentHashingRouterSpec.scala Akka example source code

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

import scala.concurrent.Await
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.Props
import akka.pattern.ask
import akka.routing.ConsistentHashingRouter.ConsistentHashable
import akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope
import akka.routing.ConsistentHashingRouter.ConsistentHashMapping
import akka.testkit.AkkaSpec
import akka.testkit._

object ConsistentHashingRouterSpec {

  val config = """
    akka.actor.deployment {
      /router1 {
        router = consistent-hashing-pool
        nr-of-instances = 3
        virtual-nodes-factor = 17
      }
      /router2 {
        router = consistent-hashing-pool
        nr-of-instances = 5
      }
    }
    """

  class Echo extends Actor {
    def receive = {
      case x: ConsistentHashableEnvelope ⇒ sender() ! s"Unexpected envelope: $x"
      case _                             ⇒ sender() ! self
    }
  }

  final case class Msg(key: Any, data: String) extends ConsistentHashable {
    override def consistentHashKey = key
  }

  final case class MsgKey(name: String)

  final case class Msg2(key: Any, data: String)
}

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class ConsistentHashingRouterSpec extends AkkaSpec(ConsistentHashingRouterSpec.config) with DefaultTimeout with ImplicitSender {
  import ConsistentHashingRouterSpec._
  implicit val ec = system.dispatcher

  val router1 = system.actorOf(FromConfig.props(Props[Echo]), "router1")

  "consistent hashing router" must {
    "create routees from configuration" in {
      val currentRoutees = Await.result(router1 ? GetRoutees, timeout.duration).asInstanceOf[Routees]
      currentRoutees.routees.size should be(3)
    }

    "select destination based on consistentHashKey of the message" in {
      router1 ! Msg("a", "A")
      val destinationA = expectMsgType[ActorRef]
      router1 ! ConsistentHashableEnvelope(message = "AA", hashKey = "a")
      expectMsg(destinationA)

      router1 ! Msg(17, "B")
      val destinationB = expectMsgType[ActorRef]
      router1 ! ConsistentHashableEnvelope(message = "BB", hashKey = 17)
      expectMsg(destinationB)

      router1 ! Msg(MsgKey("c"), "C")
      val destinationC = expectMsgType[ActorRef]
      router1 ! ConsistentHashableEnvelope(message = "CC", hashKey = MsgKey("c"))
      expectMsg(destinationC)
    }

    "select destination with defined hashMapping" in {
      def hashMapping: ConsistentHashMapping = {
        case Msg2(key, data) ⇒ key
      }
      val router2 = system.actorOf(ConsistentHashingPool(nrOfInstances = 1, hashMapping = hashMapping).
        props(Props[Echo]), "router2")

      router2 ! Msg2("a", "A")
      val destinationA = expectMsgType[ActorRef]
      router2 ! ConsistentHashableEnvelope(message = "AA", hashKey = "a")
      expectMsg(destinationA)

      router2 ! Msg2(17, "B")
      val destinationB = expectMsgType[ActorRef]
      router2 ! ConsistentHashableEnvelope(message = "BB", hashKey = 17)
      expectMsg(destinationB)

      router2 ! Msg2(MsgKey("c"), "C")
      val destinationC = expectMsgType[ActorRef]
      router2 ! ConsistentHashableEnvelope(message = "CC", hashKey = MsgKey("c"))
      expectMsg(destinationC)
    }
  }

}