Client side subscriber

ClientSubscriber.scala

  object TopicAssistant {
    case class Subscribe(topic: String)
    case object SubscribeAck
  }

  // Start this actor with name "topicAssistant" on each node in the cluster
  class TopicAssistant extends Actor {
    import TopicAssistant._

    ClusterReceptionistExtension(context.system).registerService(self)

    def receive = {
      case Subscribe(topic) ⇒
        context.watch(sender)
        DistributedPubSubExtension(context.system).mediator ! DistributedPubSubMediator.Subscribe(topic, sender)
        sender ! SubscribeAck
    }
  }

  class ClientSubscriber extends Actor {
    def initialContacts = Set(
      context.actorSelection("akka.tcp://ClusterSystem@host1:2552/user/receptionist"),
      context.actorSelection("akka.tcp://ClusterSystem@host2:2552/user/receptionist"))
    val clusterClient = context.actorOf(ClusterClient.props(initialContacts), "clusterClient")

    def scheduleTick(): Unit =
      context.system.scheduler.scheduleOnce(3.seconds, self, "tick")(context.dispatcher)

    override def preStart(): Unit = {
      super.preStart()
      scheduleTick()
    }

    val subscribeMsg =
      ClusterClient.Send("/user/topicAssistant", TopicAssistant.Subscribe("content"), localAffinity = true)

    def receive = {
      case "tick" ⇒
        clusterClient ! subscribeMsg
        scheduleTick()

      case TopicAssistant.SubscribeAck ⇒
        context.watch(sender)
        context.become(initialized, discardOld = false)

    }

    def initialized(): Actor.Receive = {
      case Terminated(ref) ⇒
        scheduleTick()
        context.unbecome()
      case "tick" ⇒ // ok
      case msg    ⇒ println("GOT IT FROM TOPIC: " + msg)
    }
  }