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

This example Akka source code file (TestConductorSpec.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, akka, concurrent, identify, implicitsender, longrunningtest, multinodeconfig, stmultinodespec, test, testconductor, testconductormultijvmnode2, testconductormultijvmspec, testconductorspec, testing, testkit, time

The TestConductorSpec.scala Akka example source code

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

import language.postfixOps
import com.typesafe.config.ConfigFactory
import akka.actor.{ Props, Actor, ActorIdentity, Identify, Deploy }
import scala.concurrent.Await
import scala.concurrent.Awaitable
import scala.concurrent.duration._
import akka.testkit.ImplicitSender
import akka.testkit.LongRunningTest
import java.net.InetSocketAddress
import java.net.InetAddress
import akka.remote.testkit.{ STMultiNodeSpec, MultiNodeSpec, MultiNodeConfig }
import akka.remote.transport.ThrottlerTransportAdapter.Direction

object TestConductorMultiJvmSpec extends MultiNodeConfig {
  commonConfig(debugConfig(on = false))

  val master = role("master")
  val slave = role("slave")

  testTransport(on = true)
}

class TestConductorMultiJvmNode1 extends TestConductorSpec
class TestConductorMultiJvmNode2 extends TestConductorSpec

class TestConductorSpec extends MultiNodeSpec(TestConductorMultiJvmSpec) with STMultiNodeSpec with ImplicitSender {

  import TestConductorMultiJvmSpec._

  def initialParticipants = 2

  lazy val echo = {
    system.actorSelection(node(master) / "user" / "echo") ! Identify(None)
    expectMsgType[ActorIdentity].ref.get
  }

  "A TestConductor" must {

    "enter a barrier" taggedAs LongRunningTest in {
      runOn(master) {
        system.actorOf(Props(new Actor {
          def receive = {
            case x ⇒ testActor ! x; sender() ! x
          }
        }).withDeploy(Deploy.local), "echo")
      }

      enterBarrier("name")
    }

    "support throttling of network connections" taggedAs LongRunningTest in {

      runOn(slave) {
        // start remote network connection so that it can be throttled
        echo ! "start"
      }

      expectMsg("start")

      runOn(master) {
        testConductor.throttle(slave, master, Direction.Send, rateMBit = 0.01).await
      }

      enterBarrier("throttled_send")

      runOn(slave) {
        for (i ← 0 to 9) echo ! i
      }

      within(0.6 seconds, 2 seconds) {
        expectMsg(500 millis, 0)
        receiveN(9) should be(1 to 9)
      }

      enterBarrier("throttled_send2")

      runOn(master) {
        testConductor.throttle(slave, master, Direction.Send, -1).await
        testConductor.throttle(slave, master, Direction.Receive, rateMBit = 0.01).await
      }

      enterBarrier("throttled_recv")

      runOn(slave) {
        for (i ← 10 to 19) echo ! i
      }

      val (min, max) =
        if (isNode(master)) (0 seconds, 500 millis)
        else (0.3 seconds, 3 seconds)

      within(min, max) {
        expectMsg(500 millis, 10)
        receiveN(9) should be(11 to 19)
      }

      enterBarrier("throttled_recv2")

      runOn(master) {
        testConductor.throttle(slave, master, Direction.Receive, -1).await
      }

      enterBarrier("after")
    }

  }

}