Package org.apache.pekko.cluster.ddata.typed.scaladsl

Class ReplicatorMessageAdapter<A,B extends ReplicatedData>

java.lang.Object
org.apache.pekko.cluster.ddata.typed.scaladsl.ReplicatorMessageAdapter<A,B>
Type Parameters:
A - Message type of the requesting actor.
B - Type of the ReplicatedData.
public class ReplicatorMessageAdapter<A,B extends ReplicatedData> extends Object
When interacting with the Replicator from an actor this class provides convenient methods that adapts the response messages to the requesting actor's message protocol.

One ReplicatorMessageAdapter instance can be used for a given ReplicatedData type, e.g. an OrSet[String]. Interaction with several Keys can be used via the same adapter but they must all be of the same ReplicatedData type. For interaction with several different ReplicatedData types, e.g. an OrSet[String] and a GCounter, an adapter can be created for each type.

For the default replicator in the DistributedData extension a ReplicatorMessageAdapter can be created with DistributedData.withReplicatorMessageAdapter.

*Warning*: ReplicatorMessageAdapter is not thread-safe and must only be used from the actor corresponding to the given ActorContext. It must not be accessed from threads other than the ordinary actor message processing thread, such as Future callbacks. It must not be shared between several actor instances.

param: context The pekko.actor.typed.scaladsl.ActorContext of the requesting actor. The ReplicatorMessageAdapter can only be used in this actor. param: replicator The replicator to interact with, typically DistributedData(system).replicator. param: unexpectedAskTimeout The timeout to use for ask operations. This should be longer than the timeout given in Replicator.WriteConsistency and Replicator.ReadConsistency. The replicator will always send a reply within those timeouts so the unexpectedAskTimeout should not occur, but for cleanup in a failure situation it must still exist. If askUpdate, askGet or askDelete takes longer then this unexpectedAskTimeout a TimeoutException will be thrown by the requesting actor and may be handled by supervision.

  • Constructor Details

  • Method Details

    • apply

      public static <A, B extends ReplicatedData> ReplicatorMessageAdapter<A,B> apply(ActorContext<A> context, ActorRef<Replicator.Command> replicator, scala.concurrent.duration.FiniteDuration unexpectedAskTimeout)

    • subscribe

      public void subscribe(Key<B> key, scala.Function1<Replicator.SubscribeResponse<B>,A> responseAdapter)
      Subscribe to changes of the given key. The Replicator.Changed and Replicator.Deleted messages from the replicator are transformed to the message protocol of the requesting actor with the given responseAdapter function.

    • unsubscribe

      public void unsubscribe(Key<B> key)
      Unsubscribe from a previous subscription of a given key.
      See Also:
      • ReplicatorMessageAdapter.subscribe

    • askUpdate

      public void askUpdate(scala.Function1<ActorRef<Replicator.UpdateResponse<B>>,Replicator.Update<B>> createRequest, scala.Function1<Replicator.UpdateResponse<B>,A> responseAdapter)
      Send a Replicator.Update request to the replicator. The Replicator.UpdateResponse message is transformed to the message protocol of the requesting actor with the given responseAdapter function.

      Note that createRequest is a function that creates the Update message from the provided ActorRef[UpdateResponse] that the the replicator will send the response message back through. Use that ActorRef[UpdateResponse] as the replyTo parameter in the Update message.

    • askGet

      public void askGet(scala.Function1<ActorRef<Replicator.GetResponse<B>>,Replicator.Get<B>> createRequest, scala.Function1<Replicator.GetResponse<B>,A> responseAdapter)
      Send a Replicator.Get request to the replicator. The Replicator.GetResponse message is transformed to the message protocol of the requesting actor with the given responseAdapter function.

      Note that createRequest is a function that creates the Get message from the provided ActorRef[GetResponse] that the the replicator will send the response message back through. Use that ActorRef[GetResponse] as the replyTo parameter in the Get message.

    • askDelete

      public void askDelete(scala.Function1<ActorRef<Replicator.DeleteResponse<B>>,Replicator.Delete<B>> createRequest, scala.Function1<Replicator.DeleteResponse<B>,A> responseAdapter)
      Send a Replicator.Delete request to the replicator. The Replicator.DeleteResponse message is transformed to the message protocol of the requesting actor with the given responseAdapter function.

      Note that createRequest is a function that creates the Delete message from the provided ActorRef[DeleteResponse] that the the replicator will send the response message back through. Use that ActorRef[DeleteResponse] as the replyTo parameter in the Delete message.

    • askReplicaCount

      public void askReplicaCount(scala.Function1<ActorRef<Replicator.ReplicaCount>,Replicator.GetReplicaCount> createRequest, scala.Function1<Replicator.ReplicaCount,A> responseAdapter)
      Send a Replicator.GetReplicaCount request to the replicator. The Replicator.ReplicaCount message is transformed to the message protocol of the requesting actor with the given responseAdapter function.

      Note that createRequest is a function that creates the GetReplicaCount message from the provided ActorRef[ReplicaCount] that the the replicator will send the response message back through. Use that ActorRef[ReplicaCount] as the replyTo parameter in the GetReplicaCount message.