+ private void executeEntityOwnershipShardOperation(final ActorRef shardActor, final Object message) {
+ Future<Object> future = datastore.getActorContext().executeOperationAsync(shardActor, message, MESSAGE_TIMEOUT);
+ future.onComplete(new OnComplete<Object>() {
+ @Override
+ public void onComplete(Throwable failure, Object response) {
+ if(failure != null) {
+ LOG.debug("Error sending message {} to {}", message, shardActor, failure);
+ } else {
+ LOG.debug("{} message to {} succeeded", message, shardActor, failure);
+ }
+ }
+ }, datastore.getActorContext().getClientDispatcher());
+ }
+
+ private void executeLocalEntityOwnershipShardOperation(final Object message) {
+ if(localEntityOwnershipShard == null) {
+ Future<ActorRef> future = datastore.getActorContext().findLocalShardAsync(ENTITY_OWNERSHIP_SHARD_NAME);
+ future.onComplete(new OnComplete<ActorRef>() {
+ @Override
+ public void onComplete(Throwable failure, ActorRef shardActor) {
+ if(failure != null) {
+ LOG.error("Failed to find local {} shard", ENTITY_OWNERSHIP_SHARD_NAME, failure);
+ } else {
+ localEntityOwnershipShard = shardActor;
+ executeEntityOwnershipShardOperation(localEntityOwnershipShard, message);
+ }
+ }
+ }, datastore.getActorContext().getClientDispatcher());
+
+ } else {
+ executeEntityOwnershipShardOperation(localEntityOwnershipShard, message);
+ }
+ }
+