//设置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 hk.edu.polyu.datamining.pamap2.actor

import java.util.concurrent.TimeUnit

import akka.actor._
import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, UnreachableMember}
import hk.edu.polyu.datamining.pamap2.Main
import hk.edu.polyu.datamining.pamap2.actor.MessageProtocol.{ComputeNodeHeartBeat, DispatcherHeartBeat, RequestNodeInfo}
import hk.edu.polyu.datamining.pamap2.actor.MessageProtocolFactory.NodeInfo
import hk.edu.polyu.datamining.pamap2.database.DatabaseHelper
import hk.edu.polyu.datamining.pamap2.utils.Lang._

import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits._
import scala.concurrent.duration.Duration


object ComputeActor {
  lazy val numberOfWorker = {
    val x = Main.config.getInt("compute.number_of_worker")
    if (x > 0) x else Runtime.getRuntime.availableProcessors()
  }
}

class ComputeActor extends CommonActor {
  val workers = mutable.Set.empty[ActorRef]
  var lastTimeDispatcherHeartBeat: Long = System.currentTimeMillis()

  override def preStart = {
    // init worker pool
    for (i <- 0 until ComputeActor.numberOfWorker) {
      log info s"start worker - $i"
      workers += context.actorOf(Props[WorkerActor])
    }
    // set repeat timer to report resources
    context.system.scheduler.schedule(Duration.Zero,
      Duration.create(ActorUtils.ReportInterval, TimeUnit.MILLISECONDS), self, ComputeNodeHeartBeat)

    //    cluster.subscribe(self,initialStateMode = InitialStateAsEvents,
    //      classOf[MemberEvent],classOf[UnreachableMember])
  }

  override def postStop = {
    SingletonActor.Dispatcher.proxy ! MessageProtocol.UnRegisterComputeNode(DatabaseHelper.clusterSeedId)
    workers.foreach(context.stop)
    workers.retain(_ => false)
  }

  override def receive: Receive = {
    case ComputeNodeHeartBeat => SingletonActor.Dispatcher.proxy ! NodeInfo.newInstance(system)
      if (lastTimeDispatcherHeartBeat < getMargin)
        workers.foreach(_ ! MessageProtocol.ReBindDispatcher)
    case DispatcherHeartBeat => lastTimeDispatcherHeartBeat = System.currentTimeMillis()
    case RequestNodeInfo => SingletonActor.Dispatcher.proxy ! NodeInfo.newInstance(context.system)

    case msg => showError(s"Unsupported msg : $msg")
  }

} 

//设置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.dazito.scala.dakkabase.exceptions

import akka.actor.Actor
import akka.actor.Actor.Receive
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.{MemberEvent, UnreachableMember}
import akka.event.Logging


class ClusterController extends Actor {

    val log = Logging(context.system, this)
    val cluster = Cluster(context.system)

    @scala.throws[Exception](classOf[Exception])
    override def preStart(): Unit = {
        cluster.subscribe(self, classOf[MemberEvent], classOf[UnreachableMember])
    }

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

    override def receive: Receive = {
        case x: MemberEvent => log.info("MemberEvent: {}", x)
        case x: UnreachableMember => log.info("UnreachableMember: {}", x)
    }
} 

//设置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 com.akkadb.actors

import akka.actor.Actor
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.{UnreachableMember, MemberEvent}
import akka.event.Logging


class ClusterController extends Actor {
  val log = Logging(context.system, this)
  val cluster = Cluster(context.system)

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

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

  override def receive = {
    case x: MemberEvent => log.info("MemberEvent: {}", x)
    case x: UnreachableMember => log.info("UnreachableMember {}: ", x)
  }
}