Cluster Client

An actor system that is not part of the cluster can communicate with actors somewhere in the cluster via this ClusterClient. The client can of course be part of another cluster. It only needs to know the location of one (or more) nodes to use as 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. Also, note it's necessary to change akka.actor.provider from akka.actor.LocalActorRefProvider to akka.remote.RemoteActorRefProvider or akka.cluster.ClusterActorRefProvider when using the cluster client.

The receptionist is supposed to be started on all nodes, or all nodes with specified role, in the cluster. The receptionist can be started with the ClusterReceptionistExtension or as an ordinary actor.

You can send messages via the ClusterClient to any actor in the cluster that is registered in the DistributedPubSubMediator used by the ClusterReceptionist. The ClusterReceptionistExtension provides methods for registration of actors that should be reachable from the client. Messages are wrapped in ClusterClient.Send, ClusterClient.SendToAll or ClusterClient.Publish.

1. ClusterClient.Send

The message will be delivered to one recipient with a matching path, if any such exists. If several entries match the path the message will be delivered to one random destination. The sender() of the message can specify that local affinity is preferred, i.e. the message is sent to an actor in the same local actor system as the used receptionist actor, if any such exists, otherwise random to any other matching entry.

2. ClusterClient.SendToAll

The message will be delivered to all recipients with a matching path.

3. ClusterClient.Publish

The message will be delivered to all recipients Actors that have been registered as subscribers to the named topic.

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 preserved as the original sender(), so the client can choose to send subsequent messages directly to the actor in the cluster.

While establishing a connection to a receptionist the ClusterClient will buffer messages and send them when the connection is established. If the buffer is full the ClusterClient will throw akka.actor.StashOverflowException, which can be handled in by the supervision strategy of the parent actor. The size of the buffer can be configured by the following stash-capacity setting of the mailbox that is used by the ClusterClient actor.

akka.contrib.cluster.client {

  # How often failure detection heartbeat messages should be sent
  heartbeat-interval = 2s
  
  # Number of potentially lost/delayed heartbeats that will be
  # accepted before considering it to be an anomaly.
  # The ClusterClient is using the akka.remote.DeadlineFailureDetector, which
  # will trigger if there are no heartbeats within the duration 
  # heartbeat-interval + acceptable-heartbeat-pause, i.e. 15 seconds with
  # the default settings.
  acceptable-heartbeat-pause = 15s

  mailbox {
    mailbox-type = "akka.dispatch.UnboundedDequeBasedMailbox"
    stash-capacity = 1000
  }
}

akka.extensions = ["akka.contrib.pattern.ClusterReceptionistExtension"]

runOn(host1) {
  val serviceA = system.actorOf(Props[Service], "serviceA")
  ClusterReceptionistExtension(system).registerService(serviceA)
}

runOn(host2, host3) {
  val serviceB = system.actorOf(Props[Service], "serviceB")
  ClusterReceptionistExtension(system).registerService(serviceB)
}

runOn(client) {
  val c = system.actorOf(ClusterClient.props(initialContacts), "client")
  c ! ClusterClient.Send("/user/serviceA", "hello", localAffinity = true)
  c ! ClusterClient.SendToAll("/user/serviceB", "hi")
}

val initialContacts = Set(
  system.actorSelection("akka.tcp://OtherSys@host1:2552/user/receptionist"),
  system.actorSelection("akka.tcp://OtherSys@host2:2552/user/receptionist"))

# Settings for the ClusterReceptionistExtension
akka.contrib.cluster.receptionist {
  # Actor name of the ClusterReceptionist actor, /user/receptionist
  name = receptionist

  # Start the receptionist on members tagged with this role.
  # All members are used if undefined or empty.
  role = ""

  # The receptionist will send this number of contact points to the client
  number-of-contacts = 3

  # The actor that tunnel response messages to the client will be stopped
  # after this time of inactivity.
  response-tunnel-receive-timeout = 30s
}

akka.extensions = ["akka.contrib.pattern.ClusterReceptionistExtension"]