//设置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 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 org.cg.spark.databroker

import akka.actor.Actor
import akka.event.Logging
import akka.actor.ActorLogging
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.MemberEvent
import org.apache.spark.sql.Row
import akka.actor.Props
import akka.contrib.pattern.DistributedPubSubMediator
import akka.contrib.pattern.DistributedPubSubExtension
import akka.contrib.pattern.DistributedPubSubMediator.Subscribe
import akka.contrib.pattern.DistributedPubSubMediator.Publish
import org.apache.spark.Logging


class ChannelProducer(name: String, channel : String) extends Actor with ActorLogging {
  @transient val cluster = Cluster(context.system)
  @transient val mediator = DistributedPubSubExtension(context.system).mediator
  //mediator ! Subscribe(channel, self)
  log.info(s"join channel cluster $channel")

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

  override def postStop(): Unit =
    cluster unsubscribe self

  def receive = {
    case ChannelProducer.Produce(topic, columns, data) => {
      log.info(s"receive data feed from channel/topic $topic")
      mediator ! Publish (ChannelUtil.clusterTopic(channel, topic), ChannelProducer.Message (this.name, topic, columns, data))
    }
    case ChannelProducer.Message(from, topic, columes, data) =>
      println (s"from $from, topic $topic, $columes(0), $data(0)) ")
  }
} 

//设置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 linguistic.utils

import akka.actor.{Actor, ActorLogging, Props}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.MemberEvent

object ClusterListener {
  def props = Props(new ClusterListener)
}

class ClusterListener extends Actor with ActorLogging {

  Cluster(context.system).subscribe(self, classOf[MemberEvent])

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

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

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

import akka.actor.Props
import akka.actor.UntypedActor
import akka.cluster.Cluster
import akka.cluster.ClusterEvent
import akka.cluster.ClusterEvent.MemberEvent
import akka.cluster.ClusterEvent.MemberUp
import akka.cluster.ClusterEvent.MemberRemoved
import akka.cluster.ClusterEvent.UnreachableMember
import akka.event.Logging
import akka.event.LoggingAdapter
import scala.concurrent.ExecutionContext.Implicits.global

object SimpleClusterListener {
  def props: Props = Props.create(classOf[SimpleClusterListener])
}

class SimpleClusterListener extends UntypedActor {

  private[peksimet] val log = Logging.getLogger(getContext.system, this)
  private[peksimet] val cluster = Cluster.get(getContext.system)

  override def preStart(): Unit = {
    cluster.subscribe(getSelf, ClusterEvent.initialStateAsEvents,
      classOf[ClusterEvent.MemberEvent], classOf[ClusterEvent.UnreachableMember])
  }

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

  override def onReceive(message: Any): Unit = {
    message match {
      case mUp: MemberUp =>
        log.info("Member is Up: {}", mUp.member)
      case _ => message match {
        case mUnreachable: UnreachableMember =>
          log.info("Member detected as unreachable: {}", mUnreachable.member)
        case _ => message match {
          case mRemoved: MemberRemoved =>
            log.info("Member is Removed: {}", mRemoved.member)
          case _: ClusterEvent.MemberEvent =>

          case _ => unhandled(message)
        }
      }
    }
  }
}