+ //if shard is local
+ ActorRef dataChangeListenerActor = actorContext.getActorSystem().actorOf(DataChangeListener.props(listener));
+ Future future = actorContext.executeOperationAsync(shard.get(),
+ new RegisterChangeListener(path, dataChangeListenerActor.path(), scope),
+ new Timeout(actorContext.getOperationDuration().$times(REGISTER_DATA_CHANGE_LISTENER_TIMEOUT_FACTOR)));
+
+ final DataChangeListenerRegistrationProxy listenerRegistrationProxy =
+ new DataChangeListenerRegistrationProxy(listener, dataChangeListenerActor);
+
+ future.onComplete(new OnComplete() {
+
+ @Override
+ public void onComplete(Throwable failure, Object result)
+ throws Throwable {
+ if (failure != null) {
+ LOG.error("Failed to register listener at path " + path.toString(), failure);
+ return;
+ }
+ RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
+ listenerRegistrationProxy.setListenerRegistrationActor(actorContext
+ .actorSelection(reply.getListenerRegistrationPath()));
+ }
+ }, actorContext.getActorSystem().dispatcher());
+
+ return listenerRegistrationProxy;