+ ActorRef shardManagerActor = datastore.getActorContext().getShardManager();
+
+ Configuration configuration = datastore.getActorContext().getConfiguration();
+ Collection<String> entityOwnersMemberNames = configuration.getUniqueMemberNamesForAllShards();
+ CreateShard createShard = new CreateShard(new ModuleShardConfiguration(EntityOwners.QNAME.getNamespace(),
+ "entity-owners", ENTITY_OWNERSHIP_SHARD_NAME, ModuleShardStrategy.NAME, entityOwnersMemberNames),
+ newShardBuilder(), null);
+
+ Future<Object> createFuture = datastore.getActorContext().executeOperationAsync(shardManagerActor,
+ createShard, MESSAGE_TIMEOUT);
+
+ createFuture.onComplete(new OnComplete<Object>() {
+ @Override
+ public void onComplete(Throwable failure, Object response) {
+ if(failure != null) {
+ LOG.error("Failed to create {} shard", ENTITY_OWNERSHIP_SHARD_NAME);
+ } else {
+ LOG.info("Successfully created {} shard", ENTITY_OWNERSHIP_SHARD_NAME);
+ }
+ }
+ }, datastore.getActorContext().getClientDispatcher());
+ }
+
+ 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);
+ }