/**

 * This actor is intended to be used on an external node that is not member

 * of the cluster. It acts like a gateway for sending messages to actors

 * somewhere in the cluster. From the initial contact points it will establish

 * a connection to a [[ClusterReceptionist]] somewhere in the cluster. It will

 * monitor the connection to the receptionist and establish a new connection if

 * the link goes down. When looking for a new receptionist it uses fresh contact

 * points retrieved from previous establishment, or periodically refreshed

 * contacts, i.e. not necessarily the initial contact points.

 *

 * You can send messages via the `ClusterClient` to any actor in the cluster

 * that is registered in the [[ClusterReceptionist]].

 * Messages are wrapped in [[ClusterClient.Send]], [[ClusterClient.SendToAll]]

 * or [[ClusterClient.Publish]].

 *

 *  Use the factory method [[ClusterClient#props]]) to create the

 * [[akka.actor.Props]] for the actor.

 *

 * If the receptionist is not currently available, the client will buffer the messages

 * and then deliver them when the connection to the receptionist has been established.

 * The size of the buffer is configurable and it can be disabled by using a buffer size

 * of 0. When the buffer is full old messages will be dropped when new messages are sent

 * via the client.

 *

 * Note that this is a best effort implementation: messages can always be lost due to the distributed

 * nature of the actors involved.

 */

final class ClusterClient(settings: ClusterClientSettings) extends Actor with ActorLogging

sendGetContacts()

scheduleRefreshContactsTick(establishingGetContactsInterval)

self ! RefreshContactsTick

def sendGetContacts(): Unit = {

    val sendTo =

      if (contacts.isEmpty) initialContactsSel

      else if (contacts.size == 1) initialContactsSel union contacts

      else contacts

    if (log.isDebugEnabled)

      log.debug(s"""Sending GetContacts to [${sendTo.mkString(",")}]""")

    sendTo.foreach { _ ! GetContacts }

  }

def scheduleRefreshContactsTick(interval: FiniteDuration): Unit = {

    refreshContactsTask foreach { _.cancel() }

    refreshContactsTask = Some(context.system.scheduler.schedule(

      interval, interval, self, RefreshContactsTick))

  }

case RefreshContactsTick ⇒ sendGetContacts()

var contactPaths: HashSet[ActorPath] =

    initialContacts.to[HashSet]

  val initialContactsSel =

    contactPaths.map(context.actorSelection)

  var contacts = initialContactsSel

case ActorIdentity(_, Some(receptionist)) ⇒

        log.info("Connected to [{}]", receptionist.path)

        scheduleRefreshContactsTick(refreshContactsInterval)

        sendBuffered(receptionist)

        context.become(active(receptionist) orElse contactPointMessages)

        connectTimerCancelable.foreach(_.cancel())

        failureDetector.heartbeat()

        self ! HeartbeatTick // will register us as active client of the selected receptionist

def active(receptionist: ActorRef): Actor.Receive = {

    case Send(path, msg, localAffinity) ⇒

      receptionist forward DistributedPubSubMediator.Send(path, msg, localAffinity)

    case SendToAll(path, msg) ⇒

      receptionist forward DistributedPubSubMediator.SendToAll(path, msg)

    case Publish(topic, msg) ⇒

      receptionist forward DistributedPubSubMediator.Publish(topic, msg)

    case HeartbeatTick ⇒

      if (!failureDetector.isAvailable) {

        log.info("Lost contact with [{}], reestablishing connection", receptionist)

        reestablish()

      } else

        receptionist ! Heartbeat

    case HeartbeatRsp ⇒

      failureDetector.heartbeat()

    case RefreshContactsTick ⇒

      receptionist ! GetContacts

    case Contacts(contactPoints) ⇒

      // refresh of contacts

      if (contactPoints.nonEmpty) {

        contactPaths = contactPoints.map(ActorPath.fromString).to[HashSet]

        contacts = contactPaths.map(context.actorSelection)

      }

      publishContactPoints()

    case _: ActorIdentity ⇒ // ok, from previous establish, already handled

    case ReceptionistShutdown ⇒

      if (receptionist == sender()) {

        log.info("Receptionist [{}] is shutting down, reestablishing connection", receptionist)

        reestablish()

      }

  }

val heartbeatTask = context.system.scheduler.schedule(

    heartbeatInterval, heartbeatInterval, self, HeartbeatTick)

object ClusterClientReceptionist extends ExtensionId[ClusterClientReceptionist] with ExtensionIdProvider {

  override def get(system: ActorSystem): ClusterClientReceptionist = super.get(system)

  override def lookup() = ClusterClientReceptionist

  override def createExtension(system: ExtendedActorSystem): ClusterClientReceptionist =

    new ClusterClientReceptionist(system)

}

/**

 * Extension that starts [[ClusterReceptionist]] and accompanying [[akka.cluster.pubsub.DistributedPubSubMediator]]

 * with settings defined in config section `akka.cluster.client.receptionist`.

 * The [[akka.cluster.pubsub.DistributedPubSubMediator]] is started by the [[akka.cluster.pubsub.DistributedPubSub]] extension.

 */

final class ClusterClientReceptionist(system: ExtendedActorSystem) extends Extension

/**

   * The [[ClusterReceptionist]] actor

   */

  private val receptionist: ActorRef = {

    if (isTerminated)

      system.deadLetters

    else {

      val name = config.getString("name")

      val dispatcher = config.getString("use-dispatcher") match {

        case "" ⇒ Dispatchers.DefaultDispatcherId

        case id ⇒ id

      }

      // important to use val mediator here to activate it outside of ClusterReceptionist constructor

      val mediator = pubSubMediator

      system.systemActorOf(ClusterReceptionist.props(mediator, ClusterReceptionistSettings(config))

        .withDispatcher(dispatcher), name)

    }

  }

  /**

   * Returns the underlying receptionist actor, particularly so that its

   * events can be observed via subscribe/unsubscribe.

   */

  def underlying: ActorRef =

    receptionist

/**

   * Register the actors that should be reachable for the clients in this [[DistributedPubSubMediator]].

   */

  private def pubSubMediator: ActorRef = DistributedPubSub(system).mediator

/**

 * [[ClusterClient]] connects to this actor to retrieve. The `ClusterReceptionist` is

 * supposed to be started on all nodes, or all nodes with specified role, in the cluster.

 * The receptionist can be started with the [[ClusterClientReceptionist]] or as an

 * ordinary actor (use the factory method [[ClusterReceptionist#props]]).

 *

 * The receptionist forwards messages from the client to the associated [[akka.cluster.pubsub.DistributedPubSubMediator]],

 * i.e. the client can send messages to any actor in the cluster that is registered in the

 * `DistributedPubSubMediator`. Messages from the client are wrapped in

 * [[akka.cluster.pubsub.DistributedPubSubMediator.Send]], [[akka.cluster.pubsub.DistributedPubSubMediator.SendToAll]]

 * or [[akka.cluster.pubsub.DistributedPubSubMediator.Publish]] with the semantics described in

 * [[akka.cluster.pubsub.DistributedPubSubMediator]].

 *

 * Response messages from the destination actor are tunneled via the receptionist

 * to avoid inbound connections from other cluster nodes to the client, i.e.

 * the `sender()`, as seen by the destination actor, is not the client itself.

 * The `sender()` of the response messages, as seen by the client, is `deadLetters`

 * since the client should normally send subsequent messages via the `ClusterClient`.

 * It is possible to pass the original sender inside the reply messages if

 * the client is supposed to communicate directly to the actor in the cluster.

 *

 */

final class ClusterReceptionist(pubSubMediator: ActorRef, settings: ClusterReceptionistSettings)

  extends Actor with ActorLogging

case GetContacts ⇒

      // Consistent hashing is used to ensure that the reply to GetContacts

      // is the same from all nodes (most of the time) and it also

      // load balances the client connections among the nodes in the cluster.

      if (numberOfContacts >= nodes.size) {

        val contacts = Contacts(nodes.map(a ⇒ self.path.toStringWithAddress(a))(collection.breakOut))

        if (log.isDebugEnabled)

          log.debug("Client [{}] gets contactPoints [{}] (all nodes)", sender().path, contacts.contactPoints.mkString(","))

        sender() ! contacts

      } else {

        // using toStringWithAddress in case the client is local, normally it is not, and

        // toStringWithAddress will use the remote address of the client

        val a = consistentHash.nodeFor(sender().path.toStringWithAddress(cluster.selfAddress))

        val slice = {

          val first = nodes.from(a).tail.take(numberOfContacts)

          if (first.size == numberOfContacts) first

          else first union nodes.take(numberOfContacts - first.size)

        }

        val contacts = Contacts(slice.map(a ⇒ self.path.toStringWithAddress(a))(collection.breakOut))

        if (log.isDebugEnabled)

          log.debug("Client [{}] gets contactPoints [{}]", sender().path, contacts.contactPoints.mkString(","))

        sender() ! contacts

      }

  case msg @ (_: Send | _: SendToAll | _: Publish) ⇒

      val tunnel = responseTunnel(sender())

      tunnel ! Ping // keep alive

      pubSubMediator.tell(msg, tunnel)

def responseTunnel(client: ActorRef): ActorRef = {

    val encName = URLEncoder.encode(client.path.toSerializationFormat, "utf-8")

    context.child(encName) match {

      case Some(tunnel) ⇒ tunnel

      case None ⇒

        context.actorOf(Props(classOf[ClientResponseTunnel], client, responseTunnelReceiveTimeout), encName)

    }

  }

/**

     * Replies are tunneled via this actor, child of the receptionist, to avoid

     * inbound connections from other cluster nodes to the client.

     */

    class ClientResponseTunnel(client: ActorRef, timeout: FiniteDuration) extends Actor with ActorLogging {

      context.setReceiveTimeout(timeout)

      private val isAsk = {

        val pathElements = client.path.elements

        pathElements.size == 2 && pathElements.head == "temp" && pathElements.tail.head.startsWith("$")

      }

      def receive = {

        case Ping ⇒ // keep alive from client

        case ReceiveTimeout ⇒

          log.debug("ClientResponseTunnel for client [{}] stopped due to inactivity", client.path)

          context stop self

        case msg ⇒

          client.tell(msg, Actor.noSender)

          if (isAsk)

            context stop self

      }

    }

case Heartbeat ⇒

      if (verboseHeartbeat) log.debug("Heartbeat from client [{}]", sender().path)

      sender() ! HeartbeatRsp

      updateClientInteractions(sender())