diff --git a/akka-cluster-tools/src/main/resources/reference.conf b/akka-cluster-tools/src/main/resources/reference.conf index fb1bd66fc1..ed792f7235 100644 --- a/akka-cluster-tools/src/main/resources/reference.conf +++ b/akka-cluster-tools/src/main/resources/reference.conf @@ -75,6 +75,7 @@ akka.cluster.client.receptionist { } # //#receptionist-ext-config +# //#cluster-client-config # Settings for the ClusterClient akka.cluster.client { # Actor paths of the ClusterReceptionist actors on the servers (cluster nodes) @@ -111,6 +112,7 @@ akka.cluster.client { # Maximum allowed buffer size is 10000. buffer-size = 1000 } +# //#cluster-client-config # //#singleton-config akka.cluster.singleton { @@ -150,4 +152,4 @@ akka.cluster.singleton-proxy { # Maximum allowed buffer size is 10000. buffer-size = 1000 } -# //#singleton-proxy-config \ No newline at end of file +# //#singleton-proxy-config diff --git a/akka-cluster-tools/src/main/scala/akka/cluster/client/ClusterClient.scala b/akka-cluster-tools/src/main/scala/akka/cluster/client/ClusterClient.scala index 30818975cd..851726b7b8 100644 --- a/akka-cluster-tools/src/main/scala/akka/cluster/client/ClusterClient.scala +++ b/akka-cluster-tools/src/main/scala/akka/cluster/client/ClusterClient.scala @@ -107,11 +107,22 @@ final class ClusterClientSettings( require(bufferSize >= 0 && bufferSize <= 10000, "bufferSize must be >= 0 and <= 10000") + /** + * Scala API + */ def withInitialContacts(initialContacts: Set[ActorPath]): ClusterClientSettings = { require(initialContacts.nonEmpty, "initialContacts must be defined") copy(initialContacts = initialContacts) } + /** + * Java API + */ + def withInitialContacts(initialContacts: java.util.Set[ActorPath]): ClusterClientSettings = { + import scala.collection.JavaConverters._ + withInitialContacts(initialContacts.asScala.toSet) + } + def withEstablishingGetContactsInterval(establishingGetContactsInterval: FiniteDuration): ClusterClientSettings = copy(establishingGetContactsInterval = establishingGetContactsInterval) diff --git a/akka-cluster-tools/src/test/java/akka/cluster/client/ClusterClientTest.java b/akka-cluster-tools/src/test/java/akka/cluster/client/ClusterClientTest.java new file mode 100644 index 0000000000..ce647c06b5 --- /dev/null +++ b/akka-cluster-tools/src/test/java/akka/cluster/client/ClusterClientTest.java @@ -0,0 +1,65 @@ +/** + * Copyright (C) 2009-2015 Typesafe Inc. + */ + +package akka.cluster.client; + +import com.typesafe.config.ConfigFactory; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import org.junit.ClassRule; +import org.junit.Test; + +import akka.actor.ActorPath; +import akka.actor.ActorPaths; +import akka.actor.ActorRef; +import akka.actor.ActorSystem; +import akka.actor.Props; +import akka.actor.UntypedActor; +import akka.testkit.AkkaJUnitActorSystemResource; + +public class ClusterClientTest { + + @ClassRule + public static AkkaJUnitActorSystemResource actorSystemResource = + new AkkaJUnitActorSystemResource("DistributedPubSubMediatorTest", + ConfigFactory.parseString( + "akka.actor.provider = \"akka.cluster.ClusterActorRefProvider\"\n" + + "akka.remote.netty.tcp.port=0")); + + private final ActorSystem system = actorSystemResource.getSystem(); + + //#initialContacts + Set initialContacts() { + return new HashSet(Arrays.asList( + ActorPaths.fromString("akka.tcp://OtherSys@host1:2552/system/receptionist"), + ActorPaths.fromString("akka.tcp://OtherSys@host2:2552/system/receptionist"))); + } + //#initialContacts + + + @Test + public void demonstrateUsage() { + //#server + ActorRef serviceA = system.actorOf(Props.create(Service.class), "serviceA"); + ClusterClientReceptionist.get(system).registerService(serviceA); + + ActorRef serviceB = system.actorOf(Props.create(Service.class), "serviceB"); + ClusterClientReceptionist.get(system).registerService(serviceB); + //#server + + //#client + final ActorRef c = system.actorOf(ClusterClient.props( + ClusterClientSettings.create(system).withInitialContacts(initialContacts())), + "client"); + c.tell(new ClusterClient.Send("/user/serviceA", "hello", true), ActorRef.noSender()); + c.tell(new ClusterClient.SendToAll("/user/serviceB", "hi"), ActorRef.noSender()); + //#client + } + + static public class Service extends UntypedActor { + public void onReceive(Object msg) { + } + } +} diff --git a/akka-docs/rst/java/cluster-client.rst b/akka-docs/rst/java/cluster-client.rst new file mode 100644 index 0000000000..90f7ec715d --- /dev/null +++ b/akka-docs/rst/java/cluster-client.rst @@ -0,0 +1,147 @@ +.. _cluster-client-java: + +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. + +.. note:: + + ``ClusterClient`` should not be used when sending messages to actors that run + within the same cluster. Similar functionality as the ``ClusterClient`` is + provided in a more efficient way by :ref:`distributed-pub-sub-java` for actors that + belong to the same cluster. + +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 ``ClusterClientReceptionist`` extension +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 ``ClusterClientReceptionist`` 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 drop old messages when new messages are sent via the client. +The size of the buffer is configurable and it can be disabled by using a buffer size of 0. + +It's worth noting that messages can always be lost because of the distributed nature +of these actors. As always, additional logic should be implemented in the destination +(acknowledgement) and in the client (retry) actors to ensure at-least-once message delivery. + +An Example +---------- + +On the cluster nodes first start the receptionist. Note, it is recommended to load the extension +when the actor system is started by defining it in the ``akka.extensions`` configuration property:: + + akka.extensions = ["akka.cluster.client.ClusterClientReceptionist"] + +Next, register the actors that should be available for the client. + +.. includecode:: ../../../akka-cluster-tools/src/test/java/akka/cluster/client/ClusterClientTest.java#server + +On the client you create the ``ClusterClient`` actor and use it as a gateway for sending +messages to the actors identified by their path (without address information) somewhere +in the cluster. + +.. includecode:: ../../../akka-cluster-tools/src/test/java/akka/cluster/client/ClusterClientTest.java#client + +The ``initialContacts`` parameter is a ``Set``, which can be created like this: + +.. includecode:: ../../../akka-cluster-tools/src/test/java/akka/cluster/client/ClusterClientTest.java#initialContacts + +You will probably define the address information of the initial contact points in configuration or system property. +See also :ref:`cluster-client-config-java`. + +A more comprehensive sample is available in the `Typesafe Activator `_ +tutorial named `Distributed workers with Akka and Java! `_. + +ClusterClientReceptionist Extension +----------------------------------- + +In the example above the receptionist is started and accessed with the ``akka.cluster.client.ClusterClientReceptionist`` extension. +That is convenient and perfectly fine in most cases, but it can be good to know that it is possible to +start the ``akka.cluster.client.ClusterReceptionist`` actor as an ordinary actor and you can have several +different receptionists at the same time, serving different types of clients. + +Note that the ``ClusterClientReceptionist`` uses the ``DistributedPubSub`` extension, which is described +in :ref:`distributed-pub-sub-java`. + +It is recommended to load the extension when the actor system is started by defining it in the +``akka.extensions`` configuration property:: + + akka.extensions = ["akka.cluster.client.ClusterClientReceptionist"] + +Dependencies +------------ + +To use the Cluster Client you must add the following dependency in your project. + +sbt:: + + "com.typesafe.akka" %% "akka-cluster-tools" % "@version@" @crossString@ + +maven:: + + + com.typesafe.akka + akka-cluster-tools_@binVersion@ + @version@ + + +.. _cluster-client-config-java: + +Configuration +------------- + +The ``ClusterClientReceptionist`` extension (or ``ClusterReceptionistSettings``) can be configured +with the following properties: + +.. includecode:: ../../../akka-cluster-tools/src/main/resources/reference.conf#receptionist-ext-config + +The following configuration properties are read by the ``ClusterClientSettings`` +when created with a ``ActorSystem`` parameter. It is also possible to amend the ``ClusterClientSettings`` +or create it from another config section with the same layout as below. ``ClusterClientSettings`` is +a parameter to the ``ClusterClient.props`` factory method, i.e. each client can be configured +with different settings if needed. + +.. includecode:: ../../../akka-cluster-tools/src/main/resources/reference.conf#cluster-client-config diff --git a/akka-docs/rst/java/cluster-usage.rst b/akka-docs/rst/java/cluster-usage.rst index c472db327d..40a093caf8 100644 --- a/akka-docs/rst/java/cluster-usage.rst +++ b/akka-docs/rst/java/cluster-usage.rst @@ -340,7 +340,7 @@ Communication from an actor system that is not part of the cluster to actors run somewhere in the cluster. The client does not have to know on which node the destination actor is running. -See :ref:`cluster-client` in the contrib module. +See :ref:`cluster-client-java`. Distributed Data ^^^^^^^^^^^^^^^^ diff --git a/akka-docs/rst/java/index-network.rst b/akka-docs/rst/java/index-network.rst index 86f34c2d03..aa8b6a5c38 100644 --- a/akka-docs/rst/java/index-network.rst +++ b/akka-docs/rst/java/index-network.rst @@ -8,7 +8,7 @@ Networking cluster-usage cluster-singleton distributed-pub-sub - ../scala/cluster-client + cluster-client cluster-sharding cluster-metrics distributed-data diff --git a/akka-docs/rst/scala/cluster-client.rst b/akka-docs/rst/scala/cluster-client.rst index 11842ad6ae..bc28e61451 100644 --- a/akka-docs/rst/scala/cluster-client.rst +++ b/akka-docs/rst/scala/cluster-client.rst @@ -1,4 +1,4 @@ -.. _cluster-client: +.. _cluster-client-scala: Cluster Client ============== @@ -10,13 +10,19 @@ contact points. It will establish a connection to a ``ClusterReceptionist`` some 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 +i.e. not necessarily the initial contact points. + +.. note:: + + ``ClusterClient`` should not be used when sending messages to actors that run + within the same cluster. Similar functionality as the ``ClusterClient`` is + provided in a more efficient way by :ref:`distributed-pub-sub-scala` for actors that + belong to the same cluster. + +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 ``ClusterClientReceptionist`` extension or as an ordinary actor. @@ -79,28 +85,24 @@ in the cluster. .. includecode:: ../../../akka-cluster-tools/src/multi-jvm/scala/akka/cluster/client/ClusterClientSpec.scala#client -The ``initialContacts`` parameter is a ``Set[ActorSelection]``, which can be created like this: +The ``initialContacts`` parameter is a ``Set[ActorPath]``, which can be created like this: .. includecode:: ../../../akka-cluster-tools/src/multi-jvm/scala/akka/cluster/client/ClusterClientSpec.scala#initialContacts You will probably define the address information of the initial contact points in configuration or system property. +See also :ref:`cluster-client-config-scala`. A more comprehensive sample is available in the `Typesafe Activator `_ -tutorial named `Distributed workers with Akka and Scala! `_ -and `Distributed workers with Akka and Java! `_. +tutorial named `Distributed workers with Akka and Scala! `_. -ClusterClientReceptionist ----------------------------- +ClusterClientReceptionist Extension +----------------------------------- -In the example above the receptionist is started and accessed with the ``akka.cluster.client.ClusterClientReceptionist``. +In the example above the receptionist is started and accessed with the ``akka.cluster.client.ClusterClientReceptionist`` extension. That is convenient and perfectly fine in most cases, but it can be good to know that it is possible to start the ``akka.cluster.client.ClusterReceptionist`` actor as an ordinary actor and you can have several different receptionists at the same time, serving different types of clients. -The ``ClusterClientReceptionist`` can be configured with the following properties: - -.. includecode:: ../../../akka-cluster-tools/src/main/resources/reference.conf#receptionist-ext-config - Note that the ``ClusterClientReceptionist`` uses the ``DistributedPubSub`` extension, which is described in :ref:`distributed-pub-sub-scala`. @@ -125,3 +127,21 @@ maven:: akka-cluster-tools_@binVersion@ @version@ + +.. _cluster-client-config-scala: + +Configuration +------------- + +The ``ClusterClientReceptionist`` extension (or ``ClusterReceptionistSettings``) can be configured +with the following properties: + +.. includecode:: ../../../akka-cluster-tools/src/main/resources/reference.conf#receptionist-ext-config + +The following configuration properties are read by the ``ClusterClientSettings`` +when created with a ``ActorSystem`` parameter. It is also possible to amend the ``ClusterClientSettings`` +or create it from another config section with the same layout as below. ``ClusterClientSettings`` is +a parameter to the ``ClusterClient.props`` factory method, i.e. each client can be configured +with different settings if needed. + +.. includecode:: ../../../akka-cluster-tools/src/main/resources/reference.conf#cluster-client-config diff --git a/akka-docs/rst/scala/cluster-usage.rst b/akka-docs/rst/scala/cluster-usage.rst index e4a12e46c5..c9de0e3488 100644 --- a/akka-docs/rst/scala/cluster-usage.rst +++ b/akka-docs/rst/scala/cluster-usage.rst @@ -334,7 +334,7 @@ Communication from an actor system that is not part of the cluster to actors run somewhere in the cluster. The client does not have to know on which node the destination actor is running. -See :ref:`cluster-client` in the contrib module. +See :ref:`cluster-client-scala`. Distributed Data ^^^^^^^^^^^^^^^^ diff --git a/akka-docs/rst/scala/code/docs/persistence/PersistenceDocSpec.scala b/akka-docs/rst/scala/code/docs/persistence/PersistenceDocSpec.scala index aac0379da3..2379298fbc 100644 --- a/akka-docs/rst/scala/code/docs/persistence/PersistenceDocSpec.scala +++ b/akka-docs/rst/scala/code/docs/persistence/PersistenceDocSpec.scala @@ -330,7 +330,6 @@ object PersistenceDocSpec { //#nested-persist-persist-caller - class MyPersistAsyncActor extends PersistentActor { override def persistenceId = "my-stable-persistence-id" @@ -351,7 +350,7 @@ object PersistenceDocSpec { persistAsync(c + "-inner-2") { inner ⇒ sender() ! inner } } } - //#nested-persistAsync-persistAsync + //#nested-persistAsync-persistAsync } //#nested-persistAsync-persistAsync-caller