//设置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 com.bob.scalatour.akka.cluster

import akka.actor._
import akka.cluster.ClusterEvent._
import akka.cluster.{Member, MemberStatus}
import com.typesafe.config.ConfigFactory
import org.codehaus.jettison.json.JSONObject

class EventInterceptor extends ClusterRoledWorker {

  @volatile var interceptedRecords: Int = 0

  val IP_PATTERN = """[^\s]+\s+\[([^\]]+)\].+"(\d+\.\d+\.\d+\.\d+)"""".stripMargin.r

  val blackIpList = Array(
    "5.9.116.101", "103.42.176.138", "123.182.148.65", "5.45.64.205",
    "27.159.226.192", "76.164.228.218", "77.79.178.186", "104.200.31.117",
    "104.200.31.32", "104.200.31.238", "123.182.129.108", "220.161.98.39",
    "59.58.152.90", "117.26.221.236", "59.58.150.110", "123.180.229.156",
    "59.60.123.239", "117.26.222.6", "117.26.220.88", "59.60.124.227",
    "142.54.161.50", "59.58.148.52", "59.58.150.85", "202.105.90.142"
  ).toSet

  def receive = {
    case MemberUp(member) =>
      println(s"Member is Up: ${member.address}")
      register(member, getCollectorPath)
    case state: CurrentClusterState =>

      
  private def checkRecord(eventCode: String, line: String): (Boolean, JSONObject) = {
    val data: JSONObject = new JSONObject()
    var isIpInBlackList = false
    IP_PATTERN.findFirstMatchIn(line).foreach { m =>
      val rawDt = m.group(1)
      val dt = rawDt
      val realIp = m.group(2)

      data.put("eventdate", dt)
      data.put("realip", realIp)
      data.put("eventcode", eventCode)
      isIpInBlackList = blackIpList.contains(realIp)
    }
    (isIpInBlackList, data)
  }
}

object EventInterceptor extends App {

  Seq("2851", "2852").foreach { port =>
    val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" + port)
      .withFallback(ConfigFactory.parseString("akka.cluster.roles = [interceptor]"))
      .withFallback(ConfigFactory.load())
    val system = ActorSystem("event-cluster-system", config)
    val processingActor = system.actorOf(Props[EventInterceptor], name = "interceptingActor")
    println("Processing Actor: " + processingActor)
  }
} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.cluster

import akka.actor.{Actor, ActorLogging}
import akka.cluster.{Metric, NodeMetrics, Cluster}
import akka.cluster.ClusterEvent._
import akka.cluster.Member
import akka.cluster.MemberStatus


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

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

  def receive: Actor.Receive = {
    case MemberUp(member) =>
      onMemberUp(member)
    case UnreachableMember(member) =>
      onUnreachableMember(member)
    case MemberRemoved(member, previousStatus) =>
      onMemberRemoved(member, previousStatus)
    case ClusterMetricsChanged(forNode) =>
      forNode collectFirst {
        case m if m.address == cluster.selfAddress =>
          log.debug("{}", filter(m.metrics))
      }
    case _: MemberEvent =>
  }

  def filter(nodeMetrics: Set[Metric]): String = {
    val filtered = nodeMetrics collect { case v if v.name != "processors" => s"${v.name}:${v.value}" }
    s"NodeMetrics[${filtered.mkString(",")}]"
  }

  def onMemberUp(member: Member) = log.info("Member {} joined cluster.", member.address)

  def onUnreachableMember(member: Member) = log.info("Member detected as unreachable: {}", member)

  def onMemberRemoved(member: Member, previousStatus: MemberStatus) = log.info("Member is Removed: {} after {}", member.address, previousStatus)

} 

//设置package包名称以及导入依赖的类
package com.crawler.core.runners
import java.util

import akka.cluster.Member
import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}

import collection.JavaConversions._


object CrawlerConfig {

  def getConfig(clusterName: String = "crawler",
                masterIp: String = "127.0.0.1",
                myIp: String = "127.0.0.1",
                role: String): Config = {

    val port = if(role.equals(CrawlerMaster.role)) 2551 else 0
    val master = s"akka.tcp://[email protected]$masterIp:2551"

    val actorRole = role match {
      case CrawlerMaster.role => role
      case CrawlerAgent.role => role
      case _ => clientRole(role)
    }

    ConfigFactory.empty()
      .withValue("akka.actor.provider", ConfigValueFactory.fromAnyRef("cluster"))
      .withValue("akka.remote.netty.tcp.hostname", ConfigValueFactory.fromAnyRef(myIp))
      .withValue("akka.remote.netty.tcp.port", ConfigValueFactory.fromAnyRef(port))
      .withValue("akka.cluster.seed-nodes", ConfigValueFactory.fromIterable(util.Arrays.asList(master)))
      .withValue("akka.cluster.roles", ConfigValueFactory.fromIterable(util.Arrays.asList(actorRole)))
      .withValue("akka.remote.maximum-payload-bytes", ConfigValueFactory.fromAnyRef("30000000 bytes"))
      .withValue("akka.remote.netty.tcp.message-frame-size", ConfigValueFactory.fromAnyRef("30000000b"))
      .withValue("akka.remote.netty.tcp.send-buffer-size", ConfigValueFactory.fromAnyRef("30000000b"))
      .withValue("akka.remote.netty.tcp.receive-buffer-size", ConfigValueFactory.fromAnyRef("30000000b"))
      .withValue("akka.remote.netty.tcp.maximum-frame-size", ConfigValueFactory.fromAnyRef("30000000b"))
      .withValue("akka.cluster.auto-down-unreachable-after", ConfigValueFactory.fromAnyRef("10s"))
      .withValue("akka.actor.warn-about-java-serializer-usage", ConfigValueFactory.fromAnyRef("false"))
  }

  def clientRole(role: String) = s"$role-${CrawlerClient.role}"

  def parseClientRole(role: String) = role.split("-")(0)

  def isCrawlerAgent(member:Member): Boolean = member.hasRole(CrawlerAgent.role)

  def isCrawlerClient(member:Member): Boolean = member.roles.head.contains(CrawlerClient.role)

} 

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

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

object ClusterListener {
  def props(): Props = Props(new ClusterListener)
  case object GetState
}

class ClusterListener extends Actor with ActorLogging {
  import ClusterListener._

  val cluster = Cluster(context.system)

  var members = Set.empty[Member]

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

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

  override def receive: Receive = {
    case MemberUp(member) =>
      members = members + member
    case UnreachableMember(_) => //ignore

    case MemberRemoved(member, _) =>
      members = members - member

    case GetState =>
      sender() ! members
  }

} 

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

import akka.actor._
import akka.cluster.{Member, Cluster}
import akka.cluster.ClusterEvent._

private[elasticloadbalancing] class ElasticLoadBalancingAwareDowning(settings: ElasticLoadBalancingAwareDowningSettings) extends Actor with ActorLogging {
  import ElasticLoadBalancingAwareDowning._

  val cluster = Cluster(context.system)

  var unreachableMember: Set[Member] = Set()

  def receive: Receive = {
    case CurrentClusterState(_, unreachable, _, _, _) =>
      unreachableMember = unreachable
    case UnreachableMember(member) =>
      unreachableMember += member
    case ReachableMember(member) =>
      unreachableMember -= member
    case MemberRemoved(member, _) =>
      unreachableMember -= member
    case UnderlyingElasticLoadBalancingAwareDowning.CheckIfNodeIsReachableAgain(address) =>
      unreachableMember.find(_.address == address) match {
        case Some(_) => self ! DownIfRemovedFromELB(address)
        case None => log.info("Member [{}] will not be downed because it become reachable again.", address)
      }
    case UnderlyingElasticLoadBalancingAwareDowning.DownCompleted(address) =>
    case msg: DownIfRemovedFromELB =>
      createDowningActor forward UnderlyingElasticLoadBalancingAwareDowning.DownIfRemovedFromELB(msg.address)
  }

  def createDowningActor: ActorRef = context.actorOf(UnderlyingElasticLoadBalancingAwareDowning.props(self, settings))

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

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

private[elasticloadbalancing] object ElasticLoadBalancingAwareDowning {
  case class DownIfRemovedFromELB(address: Address)

  def props(settings: ElasticLoadBalancingAwareDowningSettings) = Props(new ElasticLoadBalancingAwareDowning(settings))
} 

//设置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, Address, Props, RelativeActorPath}
import akka.cluster.{Cluster, Member, MemberStatus}
import akka.cluster.ClusterEvent.{ReachableMember, UnreachableMember, _}
import messages.NodeStateUpdated


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

class StateMonitorClient(user: ActorRef) extends Actor {
  val cluster = Cluster(context.system)

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

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

  override def receive: Receive = {
    case state: CurrentClusterState => state.members.foreach(m => sendNodeStateUpdate(m))
    case memberEvent: MemberEvent => sendNodeStateUpdate(memberEvent.member)
    case reachableEvent: ReachableMember => sendNodeStateUpdate(reachableEvent.member, Some("Reachable"))
    case unreachableEvent: UnreachableMember => sendNodeStateUpdate(unreachableEvent.member, Some("Unreachable"))
  }

  private def sendNodeStateUpdate(member: Member, reachable: Option[String] = None) ={
    user ! NodeStateUpdated(member.address.toString, member.roles, reachable.getOrElse(member.status.toString))
  }

} 

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

import akka.actor.Address
import akka.cluster.{MemberStatus, Member}
import akka.cluster.MemberStatus.Down

import scala.concurrent.duration.FiniteDuration

abstract class QuorumLeaderAutoDownBase(quorumRole: Option[String], quorumSize: Int, downIfOutOfQuorum: Boolean, autoDownUnreachableAfter: FiniteDuration)
  extends QuorumAwareCustomAutoDownBase(quorumSize, autoDownUnreachableAfter) {

  override def onLeaderChanged(leader: Option[Address]): Unit = {
    if (quorumRole.isEmpty && isLeader) downPendingUnreachableMembers()
  }

  override def onRoleLeaderChanged(role: String, leader: Option[Address]): Unit = {
    quorumRole.foreach { r =>
      if (r == role && isRoleLeaderOf(r)) downPendingUnreachableMembers()
    }
  }


  override def onMemberRemoved(member: Member, previousStatus: MemberStatus): Unit = {
    if (isQuorumMet(quorumRole)) {
      if (isLeaderOf(quorumRole)) {
        downPendingUnreachableMembers()
      }
    } else {
      down(selfAddress)
    }
    super.onMemberRemoved(member, previousStatus)
  }

  override def downOrAddPending(member: Member): Unit = {
    if (isLeaderOf(quorumRole)) {
      down(member.address)
      replaceMember(member.copy(Down))
    } else {
      pendingAsUnreachable(member)
    }
  }

  override def downOrAddPendingAll(members: Set[Member]): Unit = {
    if (isQuorumMetAfterDown(members, quorumRole)) {
      members.foreach(downOrAddPending)
    } else if (downIfOutOfQuorum) {
      shutdownSelf()
    }
  }
} 

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

import akka.actor.Address
import akka.cluster.Member

import scala.concurrent.duration.FiniteDuration


abstract class LeaderAutoDownRolesBase(targetRoles: Set[String], autoDownUnreachableAfter: FiniteDuration)
  extends LeaderAwareCustomAutoDownBase(autoDownUnreachableAfter){


  override def onLeaderChanged(leader: Option[Address]): Unit = {
    if (isLeader) downPendingUnreachableMembers()
  }

  override def downOrAddPending(member: Member): Unit = {
    if (targetRoles.exists(role => member.hasRole(role))) {
      if (isLeader) {
        down(member.address)
      } else {
        pendingAsUnreachable(member)
      }
    }
  }

  override def downOrAddPendingAll(members: Set[Member]): Unit = {
    members.foreach(downOrAddPending)
  }
} 

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

import akka.actor.Address
import akka.cluster.Member
import scala.concurrent.duration.FiniteDuration

abstract class RoleLeaderAutoDownRolesBase(leaderRole: String, targetRoles: Set[String], autoDownUnreachableAfter: FiniteDuration)
  extends RoleLeaderAwareCustomAutoDownBase(autoDownUnreachableAfter){


  override def onRoleLeaderChanged(role: String, leader: Option[Address]): Unit = {
    if (leaderRole == role && isRoleLeaderOf(leaderRole)) downPendingUnreachableMembers()
  }

  override def downOrAddPending(member: Member): Unit = {
    if (targetRoles.exists(role => member.hasRole(role))) {
      if (isRoleLeaderOf(leaderRole)) {
        down(member.address)
      } else {
        pendingAsUnreachable(member)
      }
    }
  }

  override def downOrAddPendingAll(members: Set[Member]): Unit = {
    members.foreach(downOrAddPending)
  }
} 

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

import akka.actor.{Actor, ActorLogging, Props}
import akka.cluster.ClusterEvent._
import akka.cluster.{Member, Cluster, MemberStatus}

import scala.collection.immutable.SortedSet

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

class ClusterMembership(cluster: Cluster) extends Actor with ActorLogging {

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

  private def evolve(clusterMembers: SortedSet[Member]): Receive = {
    case MemberUp(member) =>
      log.info("MemberUp = {}", member.address)
      context become (evolve(clusterMembers + member))

    case MemberExited(member) =>
      log.info("MemberExited = {}", member.address)

    case ReachableMember(member) =>
      log.info("ReachableMember = {}", member.address)

    case UnreachableMember(member) =>
      log.info("UnreachableMember = {}", member.address)

    case MemberRemoved(member, prev) =>
      if (prev == MemberStatus.Exiting) log.info("{} gracefully exited", member.address)
      else log.info("{} downed after being Unreachable", member.address)
      context become evolve(clusterMembers - member)

    case state: CurrentClusterState =>
      log.info("Cluster state = {}", state.members)
      context become evolve(state.members)

    case 'Members =>
      sender() ! clusterMembers.mkString(",")
  }

  override def receive = evolve(SortedSet[Member]())
} 

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

import akka.actor.{Actor, ActorLogging, ExtendedActorSystem, Extension, ExtensionId, Props}
import akka.cluster.ClusterEvent._
import akka.cluster.{Cluster, Member}

import scala.collection.immutable.Queue

class LogUnreachable(system: ExtendedActorSystem) extends Extension {

  private lazy val ref = system.actorOf(Props(new Listener))

  def start(): Unit = ref


  private class Listener extends Actor with ActorLogging {
    private val cluster = Cluster(context.system)

    private var queue = Queue.empty[ClusterDomainEvent]


    override def preStart() = {
      super.preStart()

      cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[ClusterDomainEvent])
    }

    def receive = {
      case x: ClusterDomainEvent => x match {
        case x: UnreachableMember  => onUnreachableMember(x.member); enqueue(x)
        case x: MemberEvent        => enqueue(x)
        case x: ReachabilityEvent  => enqueue(x)
        case x: LeaderChanged      => enqueue(x)
        case [email protected] => enqueue(x)
        case _                     =>
      }
    }

    def enqueue(event: ClusterDomainEvent) = {
      queue = queue enqueue event takeRight 20
    }

    def onUnreachableMember(member: Member) = {
      log.warning(s"node ${member.address} is Unreachable, cluster: ${cluster.state}, events: ${queue mkString ","}")
    }
  }
}

object LogUnreachable extends ExtensionId[LogUnreachable] {
  def createExtension(system: ExtendedActorSystem) = new LogUnreachable(system)
} 

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

}