- }
- }, getContext().dispatcher());
- }
-
- private void handleForwardedCommit(ForwardedCommitTransaction message) {
- log.info("received forwarded transaction");
- modificationToCohort.put(message.getModification(), message.getCohort());
- getSelf().forward(Persistent.create(message.getModification()), getContext());
- }
-
- private void updateSchemaContext(UpdateSchemaContext message) {
- store.onGlobalContextUpdated(message.getSchemaContext());
- }
-
- private void registerChangeListener(RegisterChangeListener registerChangeListener) {
- org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> registration =
- store.registerChangeListener(registerChangeListener.getPath(), registerChangeListener.getListener(), registerChangeListener.getScope());
- ActorRef listenerRegistration = getContext().actorOf(ListenerRegistration.props(registration));
- getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf());
- }
-
- private void createTransactionChain() {
- DOMStoreTransactionChain chain = store.createTransactionChain();
- ActorRef transactionChain = getContext().actorOf(ShardTransactionChain.props(chain));
- getSender().tell(new CreateTransactionChainReply(transactionChain.path()), getSelf());
- }
+ final ListenableFuture<Void> future = cohort.commit();
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ future.get();
+ sender.tell(new CommitTransactionReply(), self);
+ } catch (InterruptedException | ExecutionException e) {
+ log.error(e, "An exception happened when committing");
+ }
+ }
+ }, getContext().dispatcher());
+ }
+
+ private void handleForwardedCommit(ForwardedCommitTransaction message) {
+ log.info("received forwarded transaction");
+ modificationToCohort
+ .put(message.getModification(), message.getCohort());
+ getSelf().forward(Persistent.create(message.getModification()),
+ getContext());
+ }
+
+ private void updateSchemaContext(UpdateSchemaContext message) {
+ store.onGlobalContextUpdated(message.getSchemaContext());
+ }
+
+ private void registerChangeListener(
+ RegisterChangeListener registerChangeListener) {
+
+ ActorSelection dataChangeListenerPath = getContext()
+ .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
+
+ AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>
+ listener = new DataChangeListenerProxy(dataChangeListenerPath);
+
+ org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>>
+ registration =
+ store.registerChangeListener(registerChangeListener.getPath(),
+ listener, registerChangeListener.getScope());
+ ActorRef listenerRegistration =
+ getContext().actorOf(ListenerRegistration.props(registration));
+ getSender()
+ .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
+ getSelf());
+ }
+
+ private void createTransactionChain() {
+ DOMStoreTransactionChain chain = store.createTransactionChain();
+ ActorRef transactionChain =
+ getContext().actorOf(ShardTransactionChain.props(chain));
+ getSender()
+ .tell(new CreateTransactionChainReply(transactionChain.path()),
+ getSelf());
+ }