+ private void commit(Persistent message) {
+ Modification modification = (Modification) message.payload();
+ DOMStoreThreePhaseCommitCohort cohort = modificationToCohort.remove(modification);
+ if(cohort == null){
+ log.error("Could not find cohort for modification : " + modification);
+ return;
+ }
+ 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());
+ }
+