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

import language.postfixOps
import scala.concurrent.duration._
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.RootActorPath
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.CurrentClusterState
import akka.cluster.ClusterEvent.MemberUp
import akka.cluster.Member
import akka.cluster.MemberStatus
import com.typesafe.config.ConfigFactory

//#backend
class TransformationBackend extends Actor {

  val cluster = Cluster(context.system)

  // subscribe to cluster changes, MemberUp
  // re-subscribe when restart
  override def preStart(): Unit = cluster.subscribe(self, classOf[MemberUp])
  override def postStop(): Unit = cluster.unsubscribe(self)

  def receive = {
    case TransformationJob(text) => sender() ! TransformationResult(text.toUpperCase)
    case state: CurrentClusterState =>
      state.members.filter(_.status == MemberStatus.Up) foreach register
    case MemberUp(m) => register(m)
  }

  def register(member: Member): Unit =
    if (member.hasRole("frontend"))
      context.actorSelection(RootActorPath(member.address) / "user" / "frontend") !
        BackendRegistration
}
//#backend

object TransformationBackend {
  def main(args: Array[String]): Unit = {
    // Override the configuration of the port when specified as program argument
    val port = if (args.isEmpty) "0" else args(0)
    val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port=$port").
      withFallback(ConfigFactory.parseString("akka.cluster.roles = [backend]")).
      withFallback(ConfigFactory.load())

    val system = ActorSystem("ClusterSystem", config)
    system.actorOf(Props[TransformationBackend], name = "backend")
  }
} 

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

import akka.actor.{ActorPath, ActorRef, Actor}
import akka.cluster.{Member, Cluster}
import akka.cluster.ClusterEvent.{MemberEvent, UnreachableMember, MemberUp, InitialStateAsEvents}

object Registration extends Serializable

trait EventMessage extends Serializable

case class RawNginxRecord(sourceHost: String, line: String) extends EventMessage

case class NginxRecord(sourceHost: String, eventCode: String, line: String) extends EventMessage

case class FilteredRecord(sourceHost: String, eventCode: String, line: String, logDate: String, realIp: String) extends EventMessage

abstract class ClusterRoledWorker extends Actor {

  // ????Cluster??
  val cluster = Cluster(context.system)
  // ????????????
  var workers = IndexedSeq.empty[ActorRef]

  @throws[Exception](classOf[Exception])
  override def preStart(): Unit = {
    super.preStart()
    // ??????
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberUp], classOf[UnreachableMember], classOf[MemberEvent])
  }

  @throws[Exception](classOf[Exception])
  override def postStop(): Unit = {
    cluster.unsubscribe(self)
    super.postStop()
  }

  def register(member: Member, createPath: (Member) => ActorPath): Unit = {
    val actorPath = createPath(member)
    println("Actor path: " + actorPath)
    val actorSelection = context.actorSelection(actorPath)
    actorSelection ! Registration
  }

}

 

//设置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 hu.jonas.client

import akka.actor.{Actor, ActorLogging, ActorPath, ActorSystem, Props}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, MemberUp, UnreachableMember}
import akka.cluster.client.{ClusterClient, ClusterClientSettings}
import com.typesafe.config.ConfigFactory
import hu.jonas.cluster.ClusterController

object Main extends App {

  val system = ActorSystem("Cluster", ConfigFactory.load("client"))

  val client = system.actorOf(Props[Client], "Proxy")
}

class Client extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

  val initialContacts = Set(
    ActorPath.fromString("akka.tcp://[email protected]:2551/system/receptionist"))

  val setting = ClusterClientSettings(context.system).withInitialContacts(initialContacts)
  val client = context.system.actorOf(ClusterClient.props(setting), "Client")

  def send(msg: Any): Unit = {
      client ! ClusterClient.Send("/user/Controller", msg, localAffinity = true)
  }

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

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

  def receive = {
    case MemberUp(member) =>
      log.info("Member is up")
      send(ClusterController.Status)
    case msg @ _ =>
      log.info(s"Got message $msg")
  }
} 

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

import akka.actor.{Props, Actor, ActorLogging}
import akka.cluster.ClusterEvent.{MemberEvent, MemberRemoved, UnreachableMember, MemberUp}



case class ReducerRequest(data: String)
case class ReducerResponse(data: String)

object Reducers {
  def props(start_id: Int, num: Int): Props = Props(new Reducers(start_id, num))
}

class Reducers(start_id: Int, num: Int = 2) extends Actor {
  (0 until num).foreach {
    case c =>
      val id = start_id + c
      val a = context.actorOf(Props[Reducer], name=s"r$id")
      println(s"reducer actor: $a")
  }

  def receive = {
    case message =>
      println(message)
  }
}

class Reducer extends Actor with ActorLogging {
  override def preStart = {
    log.debug(Thread.currentThread.getName + " is started.")
  }


  override def postStop = {
    log.debug(Thread.currentThread.getName + " is stopped.")
  }


  def receive = {
    case MemberUp(member) =>
      log.info("Member is Up: {}", member.address)

    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)

    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)

    case ReducerRequest(data) => {
      log.info(s"reduce($data")
      sender ! ReducerResponse(data)
    }
    case _: MemberEvent => // ignore
  }
} 

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

import akka.actor.{Props, Actor, ActorLogging}
import akka.cluster.ClusterEvent.{MemberEvent, MemberRemoved, UnreachableMember, MemberUp}


case class MapperResponse(data: String)
case class MapperRequest(data: String)

object Mappers {
  def props(start_id: Int, num: Int): Props = Props(new Mappers(start_id, num))
}

class Mappers(start_id: Int, num: Int = 2) extends Actor {
  (0 until num).foreach {
    case c =>
      val id = start_id + c
      val a = context.actorOf(Props[Mapper], name=s"m$id")
      println(s"mapper actor: $a")
  }

  def receive = {
    case message =>
      println(message)
  }
}

class Mapper extends Actor with ActorLogging {


  override def preStart = {
    log.debug(Thread.currentThread.getName + " is started.")
  }


  override def postStop = {
    log.debug(Thread.currentThread.getName + " is stopped.")
  }


  def receive = {
    case MemberUp(member) =>
      log.info("Member is Up: {}", member.address)

    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)

    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)

    case MapperRequest(data) =>
      log.info(s"MapperRequest($data)")
      sender ! MapperResponse(data)

    case _: MemberEvent => // ignore
  }
} 

//设置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 com.juanjo.akka.encrypt.actors

import akka.actor._
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.{MemberEvent, MemberUp, UnreachableMember}
import com.juanjo.akka.encrypt.common.domain.SecureMessage
import com.juanjo.akka.encrypt.config.ServerConfig
import org.apache.log4j.Logger

object ServerActor {
  def props(cluster: Cluster): Props = Props(new ServerActor(cluster))

}


class ServerActor(cluster:Cluster) extends Actor with ServerConfig  {
  lazy val logger = Logger.getLogger(classOf[ServerActor])

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

  }

  def receive: Receive = {
    case SecureMessage(p) => logger.info(s"******Secure message from a client: $p")
    case MemberUp(member) => logger.info(s"******New Member is up: $member")
    case UnreachableMember(member) =>logger.info(s"*******Unreachable member: $member")
  }
} 

//设置package包名称以及导入依赖的类
package de.chasmo.chat

import akka.actor._
import akka.cluster.{Cluster, MemberStatus}
import akka.cluster.ClusterEvent.{CurrentClusterState, MemberEvent, MemberRemoved, MemberUp}

class UserListener extends Actor with ActorLogging {

  val cluster = Cluster(context.system)

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

  override def postStop(): Unit =
    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
  }

}

object UserListener {

  def props: Props = Props(classOf[UserListener])

} 

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

import akka.actor.{Actor, ActorLogging, Address}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, MemberUp}
import akka.cluster.metrics.{ClusterMetricsChanged, ClusterMetricsExtension, NodeMetrics}
import akka.cluster.metrics.StandardMetrics.Cpu
     

class PerfMon extends Actor with ActorLogging {

  val cluster = Cluster(context.system)
  val extension = ClusterMetricsExtension(context.system)

  val extResourceRole = "external-resource"
  var extResourceAddr = None: Option[Address]
     
  override def preStart(): Unit = {
    extension.subscribe(self)

    cluster.subscribe(self, initialStateMode = InitialStateAsEvents,
      classOf[MemberEvent])
  }
     
  def receive = {
    case MemberUp(member) => {
      if(member.roles.contains(extResourceRole)) {
        extResourceAddr = Some(member.address)
        log.info("Member with role '{}' is Up: {}", extResourceRole, member.address)
      }
    }

    case ClusterMetricsChanged(clusterMetrics) =>
      extResourceAddr match {
        case Some(extResourceAddr) => {
          clusterMetrics.filter(_.address == extResourceAddr) foreach { nodeMetrics => 
            logCpu(nodeMetrics)
          }
        }
        case None => // No external resource node is up.
      }
  }
     
  override def postStop(): Unit = extension.unsubscribe(self)

  def logCpu(nodeMetrics: NodeMetrics): Unit = nodeMetrics match {
    case Cpu(address, timestamp, Some(systemLoadAverage), cpuCombined, cpuStolen, processors) =>
      log.info("Address: {} Load: {} ({} processors)", address, systemLoadAverage, processors)
    case _ => log.debug("No cpu info in NodeMetrics")
  }
} 

//设置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")

}