//设置package包名称以及导入依赖的类
package akka.event

import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Address
import akka.actor.Props
import akka.testkit._

object AddressTerminatedTopicBenchSpec {

  class Subscriber(testActor: ActorRef) extends Actor {
    AddressTerminatedTopic(context.system).subscribe(self)
    testActor ! "started"

    override def postStop(): Unit = {
      AddressTerminatedTopic(context.system).unsubscribe(self)
    }

    def receive = Actor.emptyBehavior
  }
}

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class AddressTerminatedTopicBenchSpec extends AkkaSpec("akka.loglevel=INFO") {
  import AddressTerminatedTopicBenchSpec._

  "Subscribe and unsubscribe of AddressTerminated" must {

    "be quick" in {
      val sys = ActorSystem(system.name + "2", system.settings.config)
      try {
        val num = 20000

        val t1 = System.nanoTime()
        val p = Props(classOf[Subscriber], testActor)
        val subscribers = Vector.fill(num)(sys.actorOf(p))
        receiveN(num, 10.seconds)
        log.info("Starting {} actors took {} ms", num, (System.nanoTime() - t1).nanos.toMillis)

        val t2 = System.nanoTime()
        shutdown(sys, 10.seconds, verifySystemShutdown = true)
        log.info("Stopping {} actors took {} ms", num, (System.nanoTime() - t2).nanos.toMillis)
      } finally {
        if (!sys.isTerminated) shutdown(sys)
      }
    }

  }
} 

//设置package包名称以及导入依赖的类
package akka.cluster

import org.scalatest.WordSpec
import org.scalatest.Matchers
import akka.actor.Address

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class NodeMetricsSpec extends WordSpec with Matchers {

  val node1 = Address("akka.tcp", "sys", "a", 2554)
  val node2 = Address("akka.tcp", "sys", "a", 2555)

  "NodeMetrics must" must {

    "return correct result for 2 'same' nodes" in {
      (NodeMetrics(node1, 0) sameAs NodeMetrics(node1, 0)) should be(true)
    }

    "return correct result for 2 not 'same' nodes" in {
      (NodeMetrics(node1, 0) sameAs NodeMetrics(node2, 0)) should be(false)
    }

    "merge 2 NodeMetrics by most recent" in {
      val sample1 = NodeMetrics(node1, 1, Set(Metric.create("a", 10, None), Metric.create("b", 20, None)).flatten)
      val sample2 = NodeMetrics(node1, 2, Set(Metric.create("a", 11, None), Metric.create("c", 30, None)).flatten)

      val merged = sample1 merge sample2
      merged.timestamp should be(sample2.timestamp)
      merged.metric("a").map(_.value) should be(Some(11))
      merged.metric("b").map(_.value) should be(Some(20))
      merged.metric("c").map(_.value) should be(Some(30))
    }

    "not merge 2 NodeMetrics if master is more recent" in {
      val sample1 = NodeMetrics(node1, 1, Set(Metric.create("a", 10, None), Metric.create("b", 20, None)).flatten)
      val sample2 = NodeMetrics(node1, 0, Set(Metric.create("a", 11, None), Metric.create("c", 30, None)).flatten)

      val merged = sample1 merge sample2 // older and not same
      merged.timestamp should be(sample1.timestamp)
      merged.metrics should be(sample1.metrics)
    }
  }
} 

//设置package包名称以及导入依赖的类
package akka.cluster

import org.scalatest.WordSpec
import org.scalatest.Matchers
import akka.actor.Address
import akka.routing.ConsistentHash
import scala.concurrent.duration._
import scala.collection.immutable

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class HeartbeatNodeRingSpec extends WordSpec with Matchers {

  val aa = UniqueAddress(Address("akka.tcp", "sys", "aa", 2552), 1)
  val bb = UniqueAddress(Address("akka.tcp", "sys", "bb", 2552), 2)
  val cc = UniqueAddress(Address("akka.tcp", "sys", "cc", 2552), 3)
  val dd = UniqueAddress(Address("akka.tcp", "sys", "dd", 2552), 4)
  val ee = UniqueAddress(Address("akka.tcp", "sys", "ee", 2552), 5)
  val ff = UniqueAddress(Address("akka.tcp", "sys", "ff", 2552), 6)

  val nodes = Set(aa, bb, cc, dd, ee, ff)

  "A HashedNodeRing" must {

    "pick specified number of nodes as receivers" in {
      val ring = HeartbeatNodeRing(cc, nodes, Set.empty, 3)
      ring.myReceivers should be(ring.receivers(cc))

      nodes foreach { n ?
        val receivers = ring.receivers(n)
        receivers.size should be(3)
        receivers should not contain (n)
      }
    }

    "pick specified number of nodes + unreachable as receivers" in {
      val ring = HeartbeatNodeRing(cc, nodes, unreachable = Set(aa, dd, ee), monitoredByNrOfMembers = 3)
      ring.myReceivers should be(ring.receivers(cc))

      ring.receivers(aa) should be(Set(bb, cc, dd, ff)) // unreachable ee skipped
      ring.receivers(bb) should be(Set(cc, dd, ee, ff)) // unreachable aa skipped
      ring.receivers(cc) should be(Set(dd, ee, ff, bb)) // unreachable aa skipped
      ring.receivers(dd) should be(Set(ee, ff, aa, bb, cc))
      ring.receivers(ee) should be(Set(ff, aa, bb, cc))
      ring.receivers(ff) should be(Set(aa, bb, cc)) // unreachable dd and ee skipped
    }

    "pick all except own as receivers when less than total number of nodes" in {
      val expected = Set(aa, bb, dd, ee, ff)
      HeartbeatNodeRing(cc, nodes, Set.empty, 5).myReceivers should be(expected)
      HeartbeatNodeRing(cc, nodes, Set.empty, 6).myReceivers should be(expected)
      HeartbeatNodeRing(cc, nodes, Set.empty, 7).myReceivers should be(expected)
    }

    "pick none when alone" in {
      val ring = HeartbeatNodeRing(cc, Set(cc), Set.empty, 3)
      ring.myReceivers should be(Set())
    }

  }
} 

//设置package包名称以及导入依赖的类
package akka.remote.transport.netty

import akka.actor.Address
import akka.remote.transport.AssociationHandle
import akka.remote.transport.AssociationHandle.{ HandleEvent, HandleEventListener, Disassociated, InboundPayload }
import akka.remote.transport.Transport.AssociationEventListener
import akka.util.ByteString
import java.net.InetSocketAddress
import org.jboss.netty.buffer.{ ChannelBuffers, ChannelBuffer }
import org.jboss.netty.channel._
import scala.concurrent.{ Future, Promise }
import scala.util.{ Success, Failure }


private[remote] class TcpAssociationHandle(val localAddress: Address,
                                           val remoteAddress: Address,
                                           val transport: NettyTransport,
                                           private val channel: Channel)
  extends AssociationHandle {
  import transport.executionContext

  override val readHandlerPromise: Promise[HandleEventListener] = Promise()

  override def write(payload: ByteString): Boolean =
    if (channel.isWritable && channel.isOpen) {
      channel.write(ChannelBuffers.wrappedBuffer(payload.asByteBuffer))
      true
    } else false

  override def disassociate(): Unit = NettyTransport.gracefulClose(channel)
} 

//设置package包名称以及导入依赖的类
package akka.remote.transport.netty

import akka.actor.Address
import akka.remote.transport.AssociationHandle
import akka.remote.transport.AssociationHandle.{ HandleEventListener, InboundPayload }
import akka.remote.transport.Transport.AssociationEventListener
import akka.util.ByteString
import java.net.{ SocketAddress, InetAddress, InetSocketAddress }
import org.jboss.netty.buffer.{ ChannelBuffer, ChannelBuffers }
import org.jboss.netty.channel._
import scala.concurrent.{ Future, Promise }


private[remote] class UdpAssociationHandle(val localAddress: Address,
                                           val remoteAddress: Address,
                                           private val channel: Channel,
                                           private val transport: NettyTransport) extends AssociationHandle {

  override val readHandlerPromise: Promise[HandleEventListener] = Promise()

  override def write(payload: ByteString): Boolean = {
    if (!channel.isConnected)
      channel.connect(new InetSocketAddress(InetAddress.getByName(remoteAddress.host.get), remoteAddress.port.get))

    if (channel.isWritable && channel.isOpen) {
      channel.write(ChannelBuffers.wrappedBuffer(payload.asByteBuffer))
      true
    } else false
  }

  override def disassociate(): Unit = try channel.close()
  finally transport.udpConnectionTable.remove(transport.addressToSocketAddress(remoteAddress))

} 

//设置package包名称以及导入依赖的类
package akka.remote

import akka.testkit._
import scala.concurrent.duration._
import akka.actor.{ Address, ExtendedActorSystem, ActorSystem }
import com.typesafe.config.ConfigFactory
import java.nio.channels.ServerSocketChannel
import java.net.InetSocketAddress
import scala.collection.JavaConverters._

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class DaemonicSpec extends AkkaSpec {

  def addr(sys: ActorSystem, proto: String) =
    sys.asInstanceOf[ExtendedActorSystem].provider.getExternalAddressFor(Address(s"akka.$proto", "", "", 0)).get

  def unusedPort = {
    val ss = ServerSocketChannel.open().socket()
    ss.bind(new InetSocketAddress("localhost", 0))
    val port = ss.getLocalPort
    ss.close()
    port
  }

  "Remoting configured with daemonic = on" must {

    "shut down correctly after getting connection refused" in {
      // get all threads running before actor system i started
      val origThreads: Set[Thread] = Thread.getAllStackTraces().keySet().asScala.to[Set]
      // create a separate actor system that we can check the threads for
      val daemonicSystem = ActorSystem("daemonic", ConfigFactory.parseString("""
        akka.daemonic = on
        akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        akka.remote.netty.tcp.transport-class = "akka.remote.transport.netty.NettyTransport"
        akka.remote.netty.tcp.port = 0
        akka.log-dead-letters-during-shutdown = off
      """))

      val unusedAddress = addr(daemonicSystem, "tcp").copy(port = Some(unusedPort))
      val selection = daemonicSystem.actorSelection(s"${unusedAddress}/user/SomeActor")
      selection ! "whatever"
      Thread.sleep(2.seconds.dilated.toMillis)

      // get new non daemonic threads running
      val newNonDaemons: Set[Thread] = Thread.getAllStackTraces().keySet().asScala.seq.
        filter(t ? !origThreads(t) && t.isDaemon == false).to[Set]

      newNonDaemons should be(Set.empty[Thread])
      shutdown(daemonicSystem)
    }
  }
} 

//设置package包名称以及导入依赖的类
package akka.remote

import akka.testkit.AkkaSpec
import akka.actor.Address
import akka.routing.ActorSelectionRoutee
import akka.routing.ConsistentRoutee
import akka.routing.ConsistentHash

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class RemoteConsistentHashingRouterSpec extends AkkaSpec("""
    akka.actor.provider = "akka.remote.RemoteActorRefProvider" """) {

  "ConsistentHashingGroup" must {

    "use same hash ring indepenent of self address" in {
      // simulating running router on two different nodes (a1, a2) with target routees on 3 other nodes (s1, s2, s3) 
      val a1 = Address("akka.tcp", "Sys", "client1", 2552)
      val a2 = Address("akka.tcp", "Sys", "client2", 2552)
      val s1 = ActorSelectionRoutee(system.actorSelection("akka.tcp://[email protected]:2552/user/a/b"))
      val s2 = ActorSelectionRoutee(system.actorSelection("akka.tcp://[email protected]:2552/user/a/b"))
      val s3 = ActorSelectionRoutee(system.actorSelection("akka.tcp://[email protected]:2552/user/a/b"))
      val nodes1 = List(ConsistentRoutee(s1, a1), ConsistentRoutee(s2, a1), ConsistentRoutee(s3, a1))
      val nodes2 = List(ConsistentRoutee(s1, a2), ConsistentRoutee(s2, a2), ConsistentRoutee(s3, a2))
      val consistentHash1 = ConsistentHash(nodes1, 10)
      val consistentHash2 = ConsistentHash(nodes2, 10)
      val keys = List("A", "B", "C", "D", "E", "F", "G")
      val result1 = keys collect { case k ? consistentHash1.nodeFor(k).routee }
      val result2 = keys collect { case k ? consistentHash2.nodeFor(k).routee }
      result1 should be(result2)
    }

  }

} 

//设置package包名称以及导入依赖的类
package akka.contrib.pattern.protobuf

import akka.actor.{ ExtendedActorSystem, Address }
import akka.testkit.AkkaSpec
import akka.contrib.pattern.DistributedPubSubMediator._
import akka.contrib.pattern.DistributedPubSubMediator.Internal._
import akka.actor.Props
import scala.collection.immutable.TreeMap

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class DistributedPubSubMessageSerializerSpec extends AkkaSpec {

  val serializer = new DistributedPubSubMessageSerializer(system.asInstanceOf[ExtendedActorSystem])

  def checkSerialization(obj: AnyRef): Unit = {
    val blob = serializer.toBinary(obj)
    val ref = serializer.fromBinary(blob, obj.getClass)
    ref should be(obj)
  }

  " DistributedPubSubMessages" must {

    "be serializable" in {
      val address1 = Address("akka.tcp", "system", "some.host.org", 4711)
      val address2 = Address("akka.tcp", "system", "other.host.org", 4711)
      val address3 = Address("akka.tcp", "system", "some.host.org", 4712)
      val u1 = system.actorOf(Props.empty, "u1")
      val u2 = system.actorOf(Props.empty, "u2")
      val u3 = system.actorOf(Props.empty, "u3")
      val u4 = system.actorOf(Props.empty, "u4")
      checkSerialization(Status(Map(address1 -> 3, address2 -> 17, address3 -> 5)))
      checkSerialization(Delta(List(
        Bucket(address1, 3, TreeMap("/user/u1" -> ValueHolder(2, Some(u1)), "/user/u2" -> ValueHolder(3, Some(u2)))),
        Bucket(address2, 17, TreeMap("/user/u3" -> ValueHolder(17, Some(u3)))),
        Bucket(address3, 5, TreeMap("/user/u4" -> ValueHolder(4, Some(u4)), "/user/u5" -> ValueHolder(5, None))))))
      checkSerialization(Send("/user/u3", "hello", localAffinity = true))
      checkSerialization(SendToAll("/user/u3", "hello", allButSelf = true))
      checkSerialization(Publish("mytopic", "hello"))
    }
  }
} 

//设置package包名称以及导入依赖的类
package homeworkzen

import akka.Done
import akka.actor.{ActorSystem, Address, CoordinatedShutdown}
import akka.cluster.Cluster
import akka.stream.ActorMaterializer
import homeworkzen.rest.RestContext
import homeworkzen.clustering.{UserManagerSingleton, UserWorkerSharding}
import homeworkzen.domain.query.JdbcJournalReader

import scala.io.StdIn

object Main extends App {
  {
    println("Initializing server...")
    implicit val actorSystem = ActorSystem("homeworkzen")
    implicit val materializer = ActorMaterializer()
    implicit val executionContext = actorSystem.dispatcher
    implicit val userManager = UserManagerSingleton.register
    implicit val userCluster = UserWorkerSharding.register
    implicit val journalReader = new JdbcJournalReader
    //val clusterLogger = actorSystem.actorOf(Props(new ClusterLogger), "ClusterLogger")
    implicit val restContext = RestContext(userManager, userCluster)
    val httpBinding = homeworkzen.rest.Routes.bindRoutes

    CoordinatedShutdown(actorSystem).addTask(CoordinatedShutdown.PhaseServiceUnbind, "api unbind")(
      () => httpBinding.flatMap(_.unbind()).map(_ => Done)
    )

    val cluster = Cluster(actorSystem)
    if (args.length == 3) {
      // todo improve args handling
      // for testing purpose only for now
      val addr = Address(cluster.selfAddress.protocol, cluster.selfAddress.system, args(3), args(4).toInt)
      cluster.join(addr)
    }
    else
      cluster.join(cluster.selfAddress)

    println(s"Server online at http://${Config.Api.interface}:${Config.Api.port}/")
    println("Server initialization done.")
    println("Press RETURN to stop server...")
    StdIn.readLine()
    println("Stopping server...")
    CoordinatedShutdown(actorSystem).run()
    ()
  }

} 

//设置package包名称以及导入依赖的类
package com.packt.chapter7

import akka.actor.{ActorSystem, Address, Deploy, Props}
import akka.remote.RemoteScope

object RemoteActorsProgrammatically1 extends App {
  val actorSystem = ActorSystem("RemoteActorsProgramatically1")
}

object RemoteActorsProgrammatically2 extends App {
  val actorSystem = ActorSystem("RemoteActorsProgramatically2")
  println("Creating actor from RemoteActorsProgramatically2")
  val address = Address("akka.tcp", "RemoteActorsProgramatically1", "127.0.0.1", 2552) // this gives the same
  val actor = actorSystem.actorOf(Props[SimpleActor].withDeploy(Deploy(scope = RemoteScope(address))), "remoteActor")
  actor ! "Checking"
} 

//设置package包名称以及导入依赖的类
package tanukki.akka.cluster.autodown.aws.elasticloadbalancing

import akka.ConfigurationException
import akka.actor.{Props, ActorSystem, Address}
import akka.cluster.{Cluster, DowningProvider}
import tanukki.akka.cluster.autodown.{ClusterCustomDowning, LeaderAutoDownRolesBase}
import scala.concurrent.duration.FiniteDuration
import scala.collection.JavaConverters._

final class LeaderAutoDowningRolesIfUnhealthy(system: ActorSystem) extends DowningProvider {

  private def clusterSettings = Cluster(system).settings

  override def downRemovalMargin: FiniteDuration = clusterSettings.DownRemovalMargin

  override def downingActorProps: Option[Props] = {
    val roles = system.settings.config.getStringList("custom-downing.aws.elasticloadbalancing.leader-auto-downing-roles-if-unhealthy.target-roles").asScala.toSet
    clusterSettings.AutoDownUnreachableAfter match {
      case d: FiniteDuration => if (roles.isEmpty) None else Some(LeaderAutoDownRolesIfUnhealthy.props(roles, d, ElasticLoadBalancingAwareDowningSettings(system)))
      case _ =>
        throw new ConfigurationException("LeaderAutoDowningRolesIfUnhealthy downing provider selected but 'akka.cluster.auto-down-unreachable-after' not set")
    }
  }
}

private[elasticloadbalancing] class LeaderAutoDownRolesIfUnhealthy(targetRoles: Set[String], autoDownUnreachableAfter: FiniteDuration, settings: ElasticLoadBalancingAwareDowningSettings)
  extends LeaderAutoDownRolesBase(targetRoles, autoDownUnreachableAfter)
  with ClusterCustomDowning {

  val elasticLoadBalancingAwareDowning = context.actorOf(ElasticLoadBalancingAwareDowning.props(settings), "elasticLoadBalancingAwareDowning")

  override def down(node: Address): Unit = {
    elasticLoadBalancingAwareDowning ! ElasticLoadBalancingAwareDowning.DownIfRemovedFromELB(node)
  }
}

private[elasticloadbalancing] object LeaderAutoDownRolesIfUnhealthy {
  def props(targetRoles: Set[String], autoDownUnreachableAfter: FiniteDuration, settings: ElasticLoadBalancingAwareDowningSettings): Props
    = Props(new LeaderAutoDownRolesIfUnhealthy(targetRoles, autoDownUnreachableAfter, settings))
} 

//设置package包名称以及导入依赖的类
package actors

import akka.actor.{Actor, ActorRef, ActorSelection, Address, Props, RelativeActorPath, RootActorPath}
import akka.actor.Actor.Receive
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import messages._

import scala.concurrent.forkjoin.ThreadLocalRandom


object PizzaClient {
  def props(servicePath: String)(user: ActorRef) = Props(classOf[PizzaClient], servicePath, user)
}

class PizzaClient(servicePath: String, user: ActorRef) extends Actor {
  val cluster = Cluster(context.system)
  var nodes = Set.empty[Address]
  val servicePathElements = servicePath match {
    case RelativeActorPath(elements) => elements
    case _ => throw new IllegalArgumentException("servicePath [%s] is not a valid relative actor path" format servicePath)
  }

  override def preStart(): Unit = {
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberEvent], classOf[ReachabilityEvent])
  }

  override def postStop(): Unit = {
    cluster.unsubscribe(self)
  }

  override def receive: Receive = {
    case order: PizzaCustomerOrder => {
      // find service actor
      val service = findServiceActor()
      service ! order
    }
    case stop: PizzaCustomerStop => {
      // find service actor
      val service = findServiceActor()
      service ! stop
    }
    case result: PizzaCustomerBaked => {
      user ! result
    }
    case PizzaCustomerPing => sender() ! PizzaCustomerPong
    case MemberUp(m) if m.hasRole("chef") => nodes += m.address
    case ReachableMember(m) if m.hasRole("chef") => nodes += m.address
    case otherMemberEvent: MemberEvent => nodes -= otherMemberEvent.member.address
    case UnreachableMember(m) => nodes -= m.address
  }

  private def findServiceActor(): ActorSelection = {
    // for load balance, select random master proxy
    val address = nodes.toIndexedSeq(ThreadLocalRandom.current.nextInt(nodes.size))
    context.actorSelection(RootActorPath(address) / servicePathElements)
  }
} 

//设置package包名称以及导入依赖的类
package com.rbmhtechnology.eventuate.tools.test

import java.util.concurrent.atomic.AtomicInteger

import akka.actor.ActorSystem
import akka.actor.Address
import akka.actor.ExtendedActorSystem
import com.rbmhtechnology.eventuate.ReplicationConnection.DefaultRemoteSystemName
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory

import scala.collection.immutable.Seq

object AkkaSystems {

  private val akkaSystemCounter = new AtomicInteger(0)

  def akkaRemotingConfig: Config = ConfigFactory.parseString(
    """
      |akka.actor.provider = akka.remote.RemoteActorRefProvider
      |akka.remote.netty.tcp.port = 0
    """.stripMargin
  )

  def akkaTestTimeoutConfig: Config = ConfigFactory.parseString(
    s"akka.test.single-expect-default=${TestTimings.timeout.duration.toMillis}ms"
  )

  def withActorSystem[A](overrideConfig: Config = ConfigFactory.empty())(f: ActorSystem => A): A =
    withActorSystems(List(overrideConfig))(systems => f(systems.head))

  def withActorSystems[A](overrideConfigs: Seq[Config])(f: Seq[ActorSystem] => A): A = {
    import Futures.AwaitHelper

    var systems = Vector.empty[ActorSystem]
    try {
      overrideConfigs.foreach { overrideConfig =>
        val config = overrideConfig
          .withFallback(akkaTestTimeoutConfig)
          .withFallback(ConfigFactory.parseResourcesAnySyntax("application.conf"))
          .withFallback(ConfigFactory.load("test-core.conf"))
        systems = systems :+ ActorSystem(newUniqueSystemName, config)
      }
      f(systems)
    } finally {
      systems.foreach(_.terminate().await)
    }
  }

  def newUniqueSystemName: String = s"$DefaultRemoteSystemName${akkaSystemCounter.getAndIncrement()}"

  def akkaAddress(system: ActorSystem): Address = system match {
    case sys: ExtendedActorSystem => sys.provider.getDefaultAddress
  }
} 

//设置package包名称以及导入依赖的类
package com.nossin.ndb

import akka.actor.{ActorRef, ActorSystem, Address, Deploy, Props}
import akka.remote.RemoteScope
import com.nossin.ndb.messages.Messages.showAddress

import scala.concurrent.duration._



object HelloAkkaRemoting3  extends App {
  val actorSystem = ActorSystem("LookingUpActors")
  implicit val dispatcher = actorSystem.dispatcher
  val selection = actorSystem.actorSelection("akka.tcp://[email protected]:2553/user/remoteActor")
  selection ! "test"
  selection.resolveOne(3 seconds).onSuccess {
    case actorRef : ActorRef =>
    println("We got an ActorRef")
    actorRef ! "test"
  }
}

object LookingUpRemoteActors extends App {
  val actorSystem = ActorSystem("LookingUpRemoteActors")
  actorSystem.actorOf(Props[ResultPrinterActor], "remoteActor")
}

object myActorSystem extends App {
  val actorSystem = ActorSystem("MyActorSystem")
}

object RemoteActors extends App {
  val actorSystem = ActorSystem("RemoteActorsProgramatically2")
  println("Creating actor from RemoteActorsProgramatically2")
  val address = Address("akka.tcp", "MyActorSystem", "127.0.0.1", 2552)
  val actor = actorSystem.actorOf(Props[ResultPrinterActor].withDeploy(Deploy(scope = RemoteScope(address))), "remoteActor")
  actor ! showAddress
} 

//设置package包名称以及导入依赖的类
package org.cg.spark.databroker

import akka.actor.ActorLogging
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.actor.Address
import akka.cluster.ClusterEvent.MemberUp
import akka.cluster.ClusterEvent.MemberRemoved
import akka.cluster.ClusterEvent.MemberEvent
import akka.actor.Actor
import akka.cluster.MemberStatus
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Props
import scala.util.Try
import java.io.File


class ChannelCluster extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

  override def preStart(): Unit = {
    cluster.subscribe(self, classOf[ChannelCluster])
  }

  override def postStop() = {
    cluster unsubscribe (self)
  }

  var nodes = Set.empty[Address]

  def receive = {
    case state: CurrentClusterState =>
      nodes = state.members.collect {
        case m if m.status == MemberStatus.Up => m.address
      }
    case MemberUp(member) =>
      nodes += member.address
      log.info("Member is Up: {}. {} nodes in cluster",
        member.address, nodes.size)
    case MemberRemoved(member, _) =>
      nodes -= member.address
      log.info("Member is Removed: {}. {} nodes cluster",
        member.address, nodes.size)
    case _: MemberEvent => // ignore
  }
} 

//设置package包名称以及导入依赖的类
package worker

import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
import akka.actor.ActorSystem
import akka.actor.Address
import akka.actor.PoisonPill
import akka.actor.Props
import akka.actor.RootActorPath
import akka.cluster.Cluster
import akka.contrib.pattern.ClusterClient
import akka.contrib.pattern.ClusterSingletonManager

object Main extends Startup {

  def main(args: Array[String]): Unit = {

    val joinAddress = startBackend(None, "backend")
    Thread.sleep(5000)
    startBackend(Some(joinAddress), "backend")
    startWorker(joinAddress)
    Thread.sleep(5000)
    startFrontend(joinAddress)
  }

}

trait Startup {

  def systemName = "FMW"
  def workTimeout = 10.seconds

  def startBackend(joinAddressOption: Option[Address], role: String): Address = {
    val conf = ConfigFactory.parseString(s"akka.cluster.roles=[$role]").
      withFallback(ConfigFactory.load())
    val system = ActorSystem(systemName, conf)
    val joinAddress = joinAddressOption.getOrElse(Cluster(system).selfAddress)
    Cluster(system).join(joinAddress)
    system.actorOf(ClusterSingletonManager.props(Master.props(workTimeout), "active",
      PoisonPill, Some(role)), "master")
    joinAddress
  }

  def startWorker(contactAddress: akka.actor.Address): Unit = {
    val system = ActorSystem(systemName)
    val initialContacts = Set(
      system.actorSelection(RootActorPath(contactAddress) / "user" / "receptionist"))
    val clusterClient = system.actorOf(ClusterClient.props(initialContacts), "clusterClient")
    system.actorOf(Worker.props(clusterClient, Props[WorkExecutor]), "worker")
  }

  def startFrontend(joinAddress: akka.actor.Address): Unit = {
    val system = ActorSystem(systemName)
    Cluster(system).join(joinAddress)
    val frontend = system.actorOf(Props[Frontend], "frontend")
    system.actorOf(Props(classOf[WorkProducer], frontend), "producer")
    system.actorOf(Props[WorkResultConsumer], "consumer")
  }
} 

//设置package包名称以及导入依赖的类
package tanukki.akka.cluster.autodown

import akka.actor.{ActorSystem, Props, Address}
import akka.cluster.{Cluster, DowningProvider}
import scala.concurrent.duration.FiniteDuration
import scala.collection.JavaConverters._
import scala.concurrent.duration._

final class RoleLeaderAutoDowningRoles(system: ActorSystem) extends DowningProvider {

  private def clusterSettings = Cluster(system).settings

  override def downRemovalMargin: FiniteDuration = clusterSettings.DownRemovalMargin

  override def downingActorProps: Option[Props] = {
    val stableAfter = system.settings.config.getDuration("custom-downing.stable-after").toMillis millis
    val leaderRole = system.settings.config.getString("custom-downing.role-leader-auto-downing-roles.leader-role")
    val roles = system.settings.config.getStringList("custom-downing.role-leader-auto-downing-roles.target-roles").asScala.toSet
    if (roles.isEmpty) None else Some(RoleLeaderAutoDownRoles.props(leaderRole, roles, stableAfter))
  }
}


private[autodown] object RoleLeaderAutoDownRoles {
  def props(leaderRole: String, targetRoles: Set[String], autoDownUnreachableAfter: FiniteDuration): Props = Props(classOf[RoleLeaderAutoDownRoles], leaderRole, targetRoles, autoDownUnreachableAfter)
}

private[autodown] class RoleLeaderAutoDownRoles(leaderRole: String, targetRoles: Set[String], autoDownUnreachableAfter: FiniteDuration)
  extends RoleLeaderAutoDownRolesBase(leaderRole, targetRoles, autoDownUnreachableAfter) with ClusterCustomDowning {

  override def down(node: Address): Unit = {
    log.info("RoleLeader is auto-downing unreachable node [{}]", node)
    cluster.down(node)
  }
} 

//设置package包名称以及导入依赖的类
package tanukki.akka.cluster.autodown

import akka.actor.{Address, ActorLogging, Scheduler}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.ClusterDomainEvent
import scala.concurrent.duration._

trait ClusterCustomDowning extends ActorLogging { base: CustomAutoDownBase =>

  val cluster = Cluster(context.system)

  override def selfAddress: Address = cluster.selfAddress

  override def scheduler: Scheduler = {
    if (context.system.scheduler.maxFrequency < 1.second / cluster.settings.SchedulerTickDuration) {
      log.warning("CustomDowning does not use a cluster dedicated scheduler. Cluster will use a dedicated scheduler if configured " +
        "with 'akka.scheduler.tick-duration' [{} ms] >  'akka.cluster.scheduler.tick-duration' [{} ms].",
        (1000 / context.system.scheduler.maxFrequency).toInt, cluster.settings.SchedulerTickDuration.toMillis)
    }
    context.system.scheduler
  }

  override def preStart(): Unit = {
    cluster.subscribe(self, classOf[ClusterDomainEvent])
  }
  override def postStop(): Unit = {
    cluster.unsubscribe(self)
  }
} 

//设置package包名称以及导入依赖的类
package tanukki.akka.cluster.autodown

import akka.ConfigurationException
import akka.actor.{Address, Props, ActorSystem}
import akka.cluster.{Cluster, DowningProvider}
import scala.concurrent.Await
import scala.concurrent.duration._

class QuorumLeaderAutoDowning(system: ActorSystem) extends DowningProvider {

  private def clusterSettings = Cluster(system).settings

  override def downRemovalMargin: FiniteDuration = clusterSettings.DownRemovalMargin

  override def downingActorProps: Option[Props] = {
    val stableAfter = system.settings.config.getDuration("custom-downing.stable-after").toMillis millis
    val role = {
      val r = system.settings.config.getString("custom-downing.quorum-leader-auto-downing.role")
      if (r.isEmpty) None else Some(r)
    }
    val quorumSize = system.settings.config.getInt("custom-downing.quorum-leader-auto-downing.quorum-size")
    val downIfOutOfQuorum = system.settings.config.getBoolean("custom-downing.quorum-leader-auto-downing.down-if-out-of-quorum")
    val shutdownActorSystem = system.settings.config.getBoolean("custom-downing.quorum-leader-auto-downing.shutdown-actor-system-on-resolution")
    Some(QuorumLeaderAutoDown.props(role, quorumSize, downIfOutOfQuorum, shutdownActorSystem, stableAfter))
  }
}


private[autodown] object QuorumLeaderAutoDown {
  def props(quorumRole: Option[String], quorumSize: Int, downIfOutOfQuorum: Boolean, shutdownActorSystem: Boolean, autoDownUnreachableAfter: FiniteDuration): Props =
    Props(classOf[QuorumLeaderAutoDown], quorumRole, quorumSize, downIfOutOfQuorum, shutdownActorSystem, autoDownUnreachableAfter)
}

private[autodown] class QuorumLeaderAutoDown(quorumRole: Option[String], quorumSize: Int, downIfOutOfQuorum: Boolean, shutdownActorSystem: Boolean, autoDownUnreachableAfter: FiniteDuration)
  extends QuorumLeaderAutoDownBase(quorumRole, quorumSize, downIfOutOfQuorum, autoDownUnreachableAfter) with ClusterCustomDowning {

  override def down(node: Address): Unit = {
    log.info("Quorum leader is auto-downing unreachable node [{}]", node)
    cluster.down(node)
  }

  override def shutdownSelf(): Unit = {
    if (shutdownActorSystem) {
      Await.result(context.system.terminate(), 10 seconds)
    } else {
      throw new SplitBrainResolvedError("QuorumLeaderAutoDowning")
    }
  }
} 

//设置package包名称以及导入依赖的类
package tanukki.akka.cluster.autodown

import akka.actor.Address
import akka.cluster.ClusterEvent._
import scala.concurrent.duration.FiniteDuration

abstract class LeaderAwareCustomAutoDownBase(autoDownUnreachableAfter: FiniteDuration) extends CustomAutoDownBase(autoDownUnreachableAfter) {

  private var leader = false

  def onLeaderChanged(leader: Option[Address]): Unit = {}

  def isLeader: Boolean = leader

  override def receiveEvent: Receive = {
    case LeaderChanged(leaderOption) =>
      leader = leaderOption.exists(_ == selfAddress)
      onLeaderChanged(leaderOption)

    case UnreachableMember(m) => unreachableMember(m)
    case ReachableMember(m)   => remove(m)
    case MemberRemoved(m, _)  => remove(m)
  }

  override def initialize(state: CurrentClusterState): Unit = {
    leader = state.leader.exists(_ == selfAddress)
    super.initialize(state)
  }
}