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

This example Akka source code file (RestartFirstSeedNodeSpec.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, address, akka, concurrent, implicitsender, list, multinodeclusterspec, multinodespec, restartfirstseednodemultijvmnode1, restartfirstseednodemultijvmnode3, restartfirstseednodespec, test, testing, time, vector

The RestartFirstSeedNodeSpec.scala Akka example source code

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

import language.postfixOps
import scala.collection.immutable
import com.typesafe.config.ConfigFactory
import org.scalatest.BeforeAndAfter
import akka.remote.testkit.MultiNodeConfig
import akka.remote.testkit.MultiNodeSpec
import akka.testkit._
import scala.concurrent.duration._
import akka.actor.Address
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.Actor
import akka.actor.RootActorPath
import akka.cluster.MemberStatus._
import akka.actor.Deploy

object RestartFirstSeedNodeMultiJvmSpec extends MultiNodeConfig {
  val seed1 = role("seed1")
  val seed2 = role("seed2")
  val seed3 = role("seed3")

  commonConfig(debugConfig(on = false).
    withFallback(ConfigFactory.parseString("akka.cluster.auto-down-unreachable-after = 0s")).
    withFallback(MultiNodeClusterSpec.clusterConfig))
}

class RestartFirstSeedNodeMultiJvmNode1 extends RestartFirstSeedNodeSpec
class RestartFirstSeedNodeMultiJvmNode2 extends RestartFirstSeedNodeSpec
class RestartFirstSeedNodeMultiJvmNode3 extends RestartFirstSeedNodeSpec

abstract class RestartFirstSeedNodeSpec
  extends MultiNodeSpec(RestartFirstSeedNodeMultiJvmSpec)
  with MultiNodeClusterSpec with ImplicitSender {

  import RestartFirstSeedNodeMultiJvmSpec._

  @volatile var seedNode1Address: Address = _

  // use a separate ActorSystem, to be able to simulate restart
  lazy val seed1System = ActorSystem(system.name, system.settings.config)

  def missingSeed = address(seed3).copy(port = Some(61313))
  def seedNodes: immutable.IndexedSeq[Address] = Vector(seedNode1Address, seed2, seed3, missingSeed)

  lazy val restartedSeed1System = ActorSystem(system.name,
    ConfigFactory.parseString("akka.remote.netty.tcp.port=" + seedNodes.head.port.get).
      withFallback(system.settings.config))

  override def afterAll(): Unit = {
    runOn(seed1) {
      if (seed1System.isTerminated)
        shutdown(restartedSeed1System)
      else
        shutdown(seed1System)
    }
    super.afterAll()
  }

  "Cluster seed nodes" must {
    "be able to restart first seed node and join other seed nodes" taggedAs LongRunningTest in within(40 seconds) {
      // seed1System is a separate ActorSystem, to be able to simulate restart
      // we must transfer its address to seed2 and seed3
      runOn(seed2, seed3) {
        system.actorOf(Props(new Actor {
          def receive = {
            case a: Address ⇒
              seedNode1Address = a
              sender() ! "ok"
          }
        }).withDeploy(Deploy.local), name = "address-receiver")
        enterBarrier("seed1-address-receiver-ready")
      }

      runOn(seed1) {
        enterBarrier("seed1-address-receiver-ready")
        seedNode1Address = Cluster(seed1System).selfAddress
        List(seed2, seed3) foreach { r ⇒
          system.actorSelection(RootActorPath(r) / "user" / "address-receiver") ! seedNode1Address
          expectMsg(5 seconds, "ok")
        }
      }
      enterBarrier("seed1-address-transfered")

      // now we can join seed1System, seed2, seed3 together
      runOn(seed1) {
        Cluster(seed1System).joinSeedNodes(seedNodes)
        awaitAssert(Cluster(seed1System).readView.members.size should be(3))
        awaitAssert(Cluster(seed1System).readView.members.map(_.status) should be(Set(Up)))
      }
      runOn(seed2, seed3) {
        cluster.joinSeedNodes(seedNodes)
        awaitMembersUp(3)
      }
      enterBarrier("started")

      // shutdown seed1System
      runOn(seed1) {
        shutdown(seed1System, remainingOrDefault)
      }
      runOn(seed2, seed3) {
        awaitMembersUp(2, canNotBePartOfMemberRing = Set(seedNodes.head))
        awaitAssert(clusterView.unreachableMembers.map(_.address) should not contain (seedNodes.head))
      }
      enterBarrier("seed1-shutdown")

      // then start restartedSeed1System, which has the same address as seed1System
      runOn(seed1) {
        Cluster(restartedSeed1System).joinSeedNodes(seedNodes)
        awaitAssert(Cluster(restartedSeed1System).readView.members.size should be(3))
        awaitAssert(Cluster(restartedSeed1System).readView.members.map(_.status) should be(Set(Up)))
      }
      runOn(seed2, seed3) {
        awaitMembersUp(3)
      }
      enterBarrier("seed1-restarted")

    }

  }
}