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

import akka.actor.{Props, ActorSystem, RootActorPath, Actor}
import akka.actor.Actor.Receive
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.MemberUp
import com.typesafe.config.ConfigFactory


class Backend extends Actor {
  val cluster = Cluster(context.system)

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

  }

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

  override def receive: Receive = {
    case Add(num1, num2) =>
      println(s"============== I'm a backend with path ${self} and I received add operation ===============")
    case MemberUp(member) =>
      if (member.hasRole("frontend")) {
        context.actorSelection(RootActorPath(member.address) / "user" / "frontend") ! BackendRegistration
      }
  }
}

object Backend {
  def initiate(port: Int) = {
    val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port").
      withFallback(ConfigFactory.load.getConfig("Backend"))

    val system = ActorSystem("ClusterSystem", config)

    val Backend = system.actorOf(Props[Backend], name = "Backend")

  }
}

case class Add(num1: Int, num2: Int)
case object BackendRegistration 

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

import akka.actor.{Actor, ActorLogging, RootActorPath}
import akka.cluster.ClusterEvent.{CurrentClusterState, MemberUp}
import akka.cluster.{Cluster, Member, MemberStatus}


class BackendActor extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

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

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

  override def receive: Receive = {
    case state: CurrentClusterState =>
      log.info("receive current cluster state")
      state.members.filter(_.status == MemberStatus.Up).foreach(register)
    case MemberUp(member) =>
      log.info("register new member")
      register(member)
    case string: String =>
      log.info(string)
  }

  def register(member: Member): Unit = {
    if (member.hasRole("frontend")) {
      context.actorSelection(RootActorPath(member.address) / "user" / "Frontend") ! "Register"
    }
  }
} 

//设置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 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 com.evolutiongaming.cluster

import akka.TestDummyActorRef
import akka.actor.{ActorRef, Address, ChildActorPath, RootActorPath}
import akka.testkit.DefaultTimeout
import com.evolutiongaming.util.ActorSpec
import org.scalatest.{FlatSpec, Matchers, OptionValues}
import org.scalatest.concurrent.{Eventually, PatienceConfiguration, ScalaFutures}
import org.scalatest.mockito.MockitoSugar

import scala.concurrent.duration._

trait AllocationStrategySpec extends FlatSpec
  with ActorSpec
  with Matchers
  with MockitoSugar
  with OptionValues
  with ScalaFutures
  with Eventually
  with PatienceConfiguration {

  override implicit val patienceConfig: PatienceConfig = PatienceConfig(5.seconds, 500.millis)

  trait AllocationStrategyScope extends ActorScope with DefaultTimeout {

    implicit val ec = system.dispatcher

    def mockedAddressRef(addr: Address): ActorRef = {
      val rootPath = RootActorPath(addr)
      val path = new ChildActorPath(rootPath, "test")
      new TestDummyActorRef(path)
    }

    def mockedHostRef(host: String): ActorRef =
      mockedAddressRef(testAddress(host))

    def testAddress(host: String): Address = Address(
      protocol = "http",
      system = "System",
      host = host,
      port = 2552)
  }
} 

//设置package包名称以及导入依赖的类
package org.alcaudon.api

import akka.actor.{Actor, ActorLogging, ActorSelection, RootActorPath}
import akka.cluster.ClusterEvent.{CurrentClusterState, MemberUp}
import akka.cluster.{Cluster, Member, MemberStatus}
import org.alcaudon.core.{ActorConfig, DataflowGraph}

object AlcaudonClient {
  case class RegisterDataflowJob(dataflow: DataflowGraph)
  case object UnknownCoordinator
}

class AlcaudonClient extends Actor with ActorLogging with ActorConfig {

  import AlcaudonClient._

  val cluster = Cluster(context.system)

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

  var coordinator: Option[ActorSelection] = None

  def receive = receiveCoordinatorNode

  def receiveCoordinatorNode: Receive = {
    case state: CurrentClusterState =>
      val coordinator = state.members
        .filter(member =>
          member.status == MemberStatus.Up && member.hasRole("coordinator"))
        .map(getCoordinatorNodePath)
      if (coordinator.size == 1)
        context.become(receiveWithCoordinator(coordinator.head))
    case MemberUp(member) =>
      if (member.hasRole("coordinator"))
        context.become(receiveWithCoordinator(getCoordinatorNodePath(member)))
    case _ =>
      sender() ! UnknownCoordinator
  }

  def receiveWithCoordinator(coordinator: ActorSelection): Receive = {
    case request: RegisterDataflowJob =>
      coordinator ! request

  }

  def getCoordinatorNodePath(member: Member): ActorSelection =
    context.actorSelection(
      RootActorPath(member.address) / "user" / "coordinator")

}