Death watch quarantine

gistfile1.scala

import org.junit.Test
import akka.actor._
import akka.actor.Identify
import java.util.concurrent.atomic.AtomicInteger
import com.typesafe.config.{Config, ConfigFactory}
import java.io.StringReader

class ThousandActorsWatchingEachOther {

  val countUp = new AtomicInteger(0)

  @Test
  def testThousand() {
    val actorSystem1: ActorSystem = ActorSystem.create("system", getLocalHostRemotingConfig(2552))
    val count = 2000
    for (i <- 0 until count) {
      actorSystem1.actorOf(Props(new Echoer), s"$i")
    }
    Thread.sleep(5000l) // give them time to be started
    
    val actorSystem2: ActorSystem = ActorSystem.create("system", getLocalHostRemotingConfig(2553))
    for (i <- 0 until count) {
      actorSystem2.actorOf(Props(new Basher(i)), s"basher-$i")
      val wait: Int = (1000 - i) / 100
      if (wait > 0) {
        Thread.sleep(wait)
      }
    }

    Thread.sleep(360000l)
  }
  
  class Echoer extends Actor {
    override def receive = {
      case x => sender.tell(x, self)
    }
  }

  class Basher(target: Int) extends Actor {
    override def preStart() {
      context.actorSelection(s"akka.tcp://system@localhost:2552/user/$target") ! Identify()
    }

    override def receive = {
      case ActorIdentity(_, None) =>
        println("Actor not found")
      case ActorIdentity(_, Some(x)) =>
        context.watch(x)
        println(countUp.incrementAndGet())
      case Terminated(x) =>
        println("Terminated!")
    }
  }

  def getLocalHostRemotingConfig(port: Int): Config = {
    ConfigFactory.parseReader(new StringReader(
      """
        |akka {
        |  actor {
        |    provider = "akka.remote.RemoteActorRefProvider"
        |  }
        |  remote {
        |    enabled-transports = ["akka.remote.netty.tcp"]
        |    netty.tcp {
        |      hostname = "localhost"
        |      port = """.stripMargin + port + """
        |    }
        | }
        |}
      """.stripMargin))
  }
}