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

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

actor, akka, cluster, concurrent, longrunningtest, minmembersbeforeupbase, minmembersbeforeupmultijvmnode2, minmembersbeforeupspec, minmembersofrolebeforeupmultijvmnode3, minmembersofrolebeforeupspec, multinodeconfig, rolename, test, testing, time, up

The MinMembersBeforeUpSpec.scala Akka example source code

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

import com.typesafe.config.ConfigFactory
import org.scalatest.BeforeAndAfter
import scala.collection.immutable.SortedSet
import scala.concurrent.duration._
import akka.remote.testconductor.RoleName
import akka.remote.testkit.MultiNodeConfig
import akka.remote.testkit.MultiNodeSpec
import akka.testkit._
import java.util.concurrent.atomic.AtomicReference
import akka.actor.Props
import akka.actor.Actor
import akka.cluster.MemberStatus._

object MinMembersBeforeUpMultiJvmSpec extends MultiNodeConfig {
  val first = role("first")
  val second = role("second")
  val third = role("third")

  commonConfig(debugConfig(on = false).withFallback(ConfigFactory.parseString(
    "akka.cluster.min-nr-of-members = 3")).
    withFallback(MultiNodeClusterSpec.clusterConfigWithFailureDetectorPuppet))
}

object MinMembersOfRoleBeforeUpMultiJvmSpec extends MultiNodeConfig {
  val first = role("first")
  val second = role("second")
  val third = role("third")

  commonConfig(debugConfig(on = false).withFallback(ConfigFactory.parseString(
    "akka.cluster.role.backend.min-nr-of-members = 2")).
    withFallback(MultiNodeClusterSpec.clusterConfigWithFailureDetectorPuppet))

  nodeConfig(first)(
    ConfigFactory.parseString("akka.cluster.roles =[frontend]"))

  nodeConfig(second, third)(
    ConfigFactory.parseString("akka.cluster.roles =[backend]"))
}

class MinMembersBeforeUpMultiJvmNode1 extends MinMembersBeforeUpSpec
class MinMembersBeforeUpMultiJvmNode2 extends MinMembersBeforeUpSpec
class MinMembersBeforeUpMultiJvmNode3 extends MinMembersBeforeUpSpec

class MinMembersOfRoleBeforeUpMultiJvmNode1 extends MinMembersOfRoleBeforeUpSpec
class MinMembersOfRoleBeforeUpMultiJvmNode2 extends MinMembersOfRoleBeforeUpSpec
class MinMembersOfRoleBeforeUpMultiJvmNode3 extends MinMembersOfRoleBeforeUpSpec

abstract class MinMembersBeforeUpSpec extends MinMembersBeforeUpBase(MinMembersBeforeUpMultiJvmSpec) {

  override def first: RoleName = MinMembersBeforeUpMultiJvmSpec.first
  override def second: RoleName = MinMembersBeforeUpMultiJvmSpec.second
  override def third: RoleName = MinMembersBeforeUpMultiJvmSpec.third

  "Cluster leader" must {
    "wait with moving members to UP until minimum number of members have joined" taggedAs LongRunningTest in {
      testWaitMovingMembersToUp()
    }
  }
}

abstract class MinMembersOfRoleBeforeUpSpec extends MinMembersBeforeUpBase(MinMembersOfRoleBeforeUpMultiJvmSpec) {

  override def first: RoleName = MinMembersOfRoleBeforeUpMultiJvmSpec.first
  override def second: RoleName = MinMembersOfRoleBeforeUpMultiJvmSpec.second
  override def third: RoleName = MinMembersOfRoleBeforeUpMultiJvmSpec.third

  "Cluster leader" must {
    "wait with moving members to UP until minimum number of members with specific role have joined" taggedAs LongRunningTest in {
      testWaitMovingMembersToUp()
    }
  }
}

abstract class MinMembersBeforeUpBase(multiNodeConfig: MultiNodeConfig)
  extends MultiNodeSpec(multiNodeConfig)
  with MultiNodeClusterSpec {

  import ClusterEvent._

  def first: RoleName
  def second: RoleName
  def third: RoleName

  def testWaitMovingMembersToUp(): Unit = {
    val onUpLatch = TestLatch(1)
    cluster.registerOnMemberUp(onUpLatch.countDown())

    runOn(first) {
      cluster join myself
      awaitAssert {
        clusterView.refreshCurrentState()
        clusterView.status should be(Joining)
      }
    }
    enterBarrier("first-started")

    onUpLatch.isOpen should be(false)

    runOn(second) {
      cluster.join(first)
    }
    runOn(first, second) {
      val expectedAddresses = Set(first, second) map address
      awaitAssert {
        clusterView.refreshCurrentState()
        clusterView.members.map(_.address) should be(expectedAddresses)
      }
      clusterView.members.map(_.status) should be(Set(Joining))
      // and it should not change
      1 to 5 foreach { _ ⇒
        Thread.sleep(1000)
        clusterView.members.map(_.address) should be(expectedAddresses)
        clusterView.members.map(_.status) should be(Set(Joining))
      }
    }
    enterBarrier("second-joined")

    runOn(third) {
      cluster.join(first)
    }
    awaitClusterUp(first, second, third)

    onUpLatch.await

    enterBarrier("after-1")
  }

}