package cn.wj.rpc

import akka.actor.{Actor, ActorSystem}
import akka.actor.Actor.Receive
import com.typesafe.config.ConfigFactory
import akka.actor.Props
import scala.concurrent.duration._

import scala.collection.mutable

/**
  * Created by WJ on 2016/12/23.
  */
class Master(val host:String,val port:Int ) extends Actor {

  // workerId -> WorkerInfo
  val idToWorker = new mutable.HashMap[String,WorkerInfo]()
  val workers = new mutable.HashSet[WorkerInfo]()

  //时间间隔时间,超时检测的间隔
  val CHECK_INTERVAL = 15000
  //用于接收消息
  override def receive: Receive = {
    case RegisterWorker(id,memory,cores) => {
//      println("a client connected")
//      sender ! "reply"  //往发送给他消息的人回复一个消息
      //判断一下是不是已经注册过了
    if(!(idToWorker.contains(id))){
      //把Worker的信息封装以前，保存到内存当中
      val workerInfo = new WorkerInfo(id,memory,cores)
      idToWorker(id) = workerInfo  //这个应该是scala的特定版本
      workers += workerInfo
      sender ! RegisteredWorker(s"akka.tcp://MasterSystem@$host:$port/user/Master")
    }

    }
    case Heartbeat(id) =>{
      if(idToWorker.contains(id)) {
        val workerInfo = idToWorker(id)
        //报活
        //得到系统当前时间
        val currentTime = System.currentTimeMillis()
        workerInfo.lastHeartbeatTime = currentTime
      }
    }

    case CheckTimeOutWorker => {
      val currentTime = System.currentTimeMillis()
      val toRemove = workers.filter(x => currentTime - x.lastHeartbeatTime > CHECK_INTERVAL)
      for(w <- toRemove){
        workers -= w
        idToWorker -= w.id
      }
      println(workers.size)
    }
  }

  override def preStart(): Unit = {
     println("prestart invoked")
     //导入隐式转换的功能
     import context.dispatcher
     context.system.scheduler.schedule(0 millis,CHECK_INTERVAL millis,self,CheckTimeOutWorker)
  }
}

object Master{
  def main(args: Array[String]): Unit = {
    val host = args(0)
    val port = args(1).toInt
    // 准备配置
    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
       """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    //ActorSystem老大,辅助创建和监控下面的Actor,他是单例的
    val actorSystem = ActorSystem("MasterSystem",config )
    //创建Actor
    val master = actorSystem.actorOf(Props(new Master(host,port)),"Master")
    actorSystem.awaitTermination()
  }
}

package cn.wj.rpc

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
/**
  * Created by WJ on 2016/12/23.
  */
class Worker(val masterHost:String,val masterPort:Int,val memory:Int,val cores:Int) extends Actor {

  var master : ActorSelection = _
  val workerId = UUID.randomUUID().toString
  val HEART_INTERVAL = 10000
  //preStart执行方法的时机:构造器之后，receive之前
  //与Master(Actor)建立连接
  override def preStart(): Unit = {
    //master已经是别的Master的引用了 ,这是跟master建立连接
    master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")
    //向Master发送注册消息
    master ! RegisterWorker(workerId,memory,cores)
  }

  override def receive: Receive = {

    case RegisteredWorker(masterUrl) => {
      println(masterUrl)
      //启动定时器发送心跳
      import context.dispatcher
      context.system.scheduler.schedule(0 millis,HEART_INTERVAL millis,self,SendHeartbeat)
    }

    case SendHeartbeat =>{
      println("send heartbeat to master")
      master ! Heartbeat(workerId)
    }

  }
}

object Worker{
  def main(args: Array[String]): Unit = {
    val host = args(0)
    val port = args(1).toInt
    val masterHost = args(2)
    val masterPort = args(3).toInt
    val memory = args(4).toInt
    val cores = args(5).toInt

    // 准备配置
    val configStr =
    s"""
       |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname = "$host"
       |akka.remote.netty.tcp.port = "$port"
       """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    //ActorSystem老大,辅助创建和监控下面的Actor,他是单例的
    val actorSystem = ActorSystem("WorkerSystem",config )
    //创建Actor,此时调用该(Actor)的prestart以及receive方法
      actorSystem.actorOf(Props(new Worker(masterHost,masterPort,memory,cores)),"Worker")
     actorSystem.awaitTermination()
  }
}

package cn.wj.rpc

/**
  * Created by WJ on 2016/12/25.
  */
trait RemoteMessage extends Serializable

//Worker->Master(这个表明当master接受这个worker时的信息，是receive)
 case class RegisterWorker (id:String, memory:Int, cores:Int) extends RemoteMessage

//Master -> Worker(这个是master收到workerd的注册信息,表明已经注册过这条信息，是sender ! xxx时候出现的)
case class RegisteredWorker(masterUrl:String) extends RemoteMessage

//这是进程之间自己给自己发送消息，所以采用case object，并且不需要实现Serializable
//Worker -> Worker(self)
case object SendHeartbeat

//这个是work向master发送定时器，其中的id是work的id，因为要向master说明，是哪一个work给他发送的心跳
//Worker -> Master
case class Heartbeat(id:String)  extends RemoteMessage

//Master -> self
case object CheckTimeOutWorker

package cn.wj.rpc

/**
  * Created by WJ on 2016/12/25.
  */
class WorkerInfo(val id:String ,val memory :Int,val cores:Int) {
    //TODO 上一次心跳
  var lastHeartbeatTime:Long = _
}