package clustertest.actor

import akka.actor.{Terminated, Actor, ActorRef, RootActorPath}
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.MemberUp

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

case class TellFriends()
case class FindFriend()
case class RegisterFriend()
case class Message(text: String)

class MessageActor extends Actor {
  val cluster = Cluster(context.system)

  var friends = IndexedSeq.empty[ActorRef]

  override def preStart() = cluster.subscribe(self, classOf[MemberUp])

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

  context.system.scheduler.schedule(Duration.Zero, 1.seconds, self, TellFriends())

  def receive = {
    case Message(text) =>
      println(text)

    case TellFriends() =>
      friends.foreach(_ ! Message("Hai from " + cluster.selfAddress.port.get))

    case FindFriend() => sender() ! RegisterFriend()

    case RegisterFriend() =>
      if (sender() != self) {
        context.watch(sender())
        friends = friends :+ sender()
      }

    case MemberUp(friend) =>
      context.actorSelection(RootActorPath(friend.address) / "user" / "message") ! FindFriend()

    case Terminated(friend) =>
      println(friend.path + " terminated!")
      friends = friends.filterNot(_ == friend)
  }
}