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

Class ReplicatorMessageAdapter<A,B extends ReplicatedData>

java.lang.Object

org.apache.pekko.cluster.ddata.typed.javadsl.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. 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 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 CompletionStage callbacks. It must not be shared between several actor instances.

param: context The pekko.actor.typed.javadsl.ActorContext of the requesting actor. The ReplicatorMessageAdapter can only be used in this actor. param: replicator The replicator to interact with, typically DistributedData.get(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 Summary

Constructors

Constructor	Description
ReplicatorMessageAdapter(ActorContext<A> context, ActorRef<Replicator.Command> replicator, Duration unexpectedAskTimeout)

Method Summary

Modifier and Type	Method	Description
void	askDelete(Function<ActorRef<Replicator.DeleteResponse<B>>,Replicator.Delete<B>> createRequest, Function<Replicator.DeleteResponse<B>,A> responseAdapter)	Send a Replicator.Delete request to the replicator.
void	askGet(Function<ActorRef<Replicator.GetResponse<B>>,Replicator.Get<B>> createRequest, Function<Replicator.GetResponse<B>,A> responseAdapter)	Send a Replicator.Get request to the replicator.
void	askReplicaCount(Function<ActorRef<Replicator.ReplicaCount>,Replicator.GetReplicaCount> createRequest, Function<Replicator.ReplicaCount,A> responseAdapter)	Send a Replicator.GetReplicaCount request to the replicator.
void	askUpdate(Function<ActorRef<Replicator.UpdateResponse<B>>,Replicator.Update<B>> createRequest, Function<Replicator.UpdateResponse<B>,A> responseAdapter)	Send a Replicator.Update request to the replicator.
void	subscribe(Key<B> key, Function<Replicator.SubscribeResponse<B>,A> responseAdapter)	Subscribe to changes of the given key.
void	unsubscribe(Key<B> key)	Unsubscribe from a previous subscription of a given key.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

ReplicatorMessageAdapter

public ReplicatorMessageAdapter(ActorContext<A> context,
 ActorRef<Replicator.Command> replicator,
 Duration unexpectedAskTimeout)

Method Details

askDelete

public void askDelete(Function<ActorRef<Replicator.DeleteResponse<B>>,Replicator.Delete<B>> createRequest,
 Function<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.

askGet

public void askGet(Function<ActorRef<Replicator.GetResponse<B>>,Replicator.Get<B>> createRequest,
 Function<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.

askReplicaCount

public void askReplicaCount(Function<ActorRef<Replicator.ReplicaCount>,Replicator.GetReplicaCount> createRequest,
 Function<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.

askUpdate

public void askUpdate(Function<ActorRef<Replicator.UpdateResponse<B>>,Replicator.Update<B>> createRequest,
 Function<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.

subscribe

public void subscribe(Key<B> key,
 Function<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