Why is my router not receiving check-in events?

Question

I'm starting to explore the new Akka Typed API. I'm trying to run an updated version of the random router from this blog post.

My router is largely the same:

import java.util.concurrent.ThreadLocalRandom

import akka.actor.Address
import akka.actor.typed.{ActorRef, Behavior}
import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
import akka.actor.typed.scaladsl.Behaviors
import akka.cluster.ClusterEvent.{ReachabilityEvent, ReachableMember, UnreachableMember}
import akka.cluster.typed.{Cluster, Subscribe}

object RandomRouter {
  private final case class WrappedReachabilityEvent(event: ReachabilityEvent)

  // subscribes to cluster reachability events and
  // avoids routees that are unreachable
  def clusterRouter[T](serviceKey: ServiceKey[T]): Behavior[T] =
    Behaviors.setup[Any] { ctx ⇒
      ctx.system.receptionist ! Receptionist.Subscribe(serviceKey, ctx.self)

      val cluster = Cluster(ctx.system)
      // typically you have to map such external messages into this
      // actor's protocol with a message adapter
      val reachabilityAdapter: ActorRef[ReachabilityEvent] = ctx.messageAdapter(WrappedReachabilityEvent.apply)

      cluster.subscriptions ! Subscribe(reachabilityAdapter, classOf[ReachabilityEvent])

      def routingBehavior(routees: Vector[ActorRef[T]], unreachable: Set[Address]): Behavior[Any] =
        Behaviors.receive { (ctx, msg) ⇒
          msg match {
            case serviceKey.Listing(services) ⇒
              if (services.isEmpty) {
                ctx.log.info("Found no services")
              } else {
                ctx.log.info(s"Found services: ${services.map(_.path.name).mkString(", ")}")
              }
              routingBehavior(services.toVector, unreachable)
            case WrappedReachabilityEvent(event) => event match {
              case UnreachableMember(m) =>
                ctx.log.warning(s"Member ${m.address} has become unreachable")
                routingBehavior(routees, unreachable + m.address)
              case ReachableMember(m) =>
                ctx.log.info(s"Member ${m.address} has become reachable again")
                routingBehavior(routees, unreachable - m.address)
            }

            case other: T @unchecked ⇒
              if (routees.isEmpty)
                Behaviors.unhandled
              else {
                val reachableRoutes =
                  if (unreachable.isEmpty) routees
                  else routees.filterNot { r => unreachable(r.path.address) }

                val i = ThreadLocalRandom.current.nextInt(reachableRoutes.size)
                reachableRoutes(i) ! other
                Behaviors.same
              }
          }
        }

      routingBehavior(Vector.empty, Set.empty)
    }.narrow[T]
}

And my cluster spins off dummy actors:

object DummyActor {
  def behavior[T](serviceKey: ServiceKey[T]): Behavior[Any] = Behaviors.setup { ctx =>

    ctx.log.info("Woohoo, I'm alive!")
    Behaviors.empty
  }
}

with the following:

object MyCluster {

  val serviceKey: ServiceKey[String] = ServiceKey[String]("cluster")

  val behavior: Behavior[String] = Behaviors.setup { ctx =>

    (1 to 5).foreach { i =>
      ctx.log.info("I'm so sleepy...")
      Thread.sleep(500)
      ctx.log.info(s"Spawning actor #$i")
      ctx.spawnAnonymous(DummyActor.behavior(serviceKey))
      ctx.log.info("I'm tired again...")
      Thread.sleep(500)
    }

    val router = ctx.spawn(RandomRouter.clusterRouter(serviceKey), "router")

    Behaviors.stopped
  }
}

When I run the following main, I always see "Found no services" in my log, indicating what I believe to mean that none of my dummy actors registered with the cluster receptionist.

import akka.actor.typed.ActorSystem

object Main extends App {

  val system = ActorSystem(MyCluster.behavior, "cluster-system")

}

What am I missing? I'm using Akka 2.5.12.

Answer 1

The dummy actors need to register! It doesn't happen automatically. This was solved by adding the following line in the setup block:

ctx.system.receptionist ! Receptionist.Register(serviceKey, ctx.self)

object DummyActor {
  def behavior[T](serviceKey: ServiceKey[T]): Behavior[Any] = Behaviors.setup { ctx =>

    ctx.system.receptionist ! Receptionist.Register(serviceKey, ctx.self)

    ctx.log.info("Woohoo, I'm alive!")
    Behaviors.empty
  }
}