opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/SingleCommitCohortProxy.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import akka.dispatch.OnComplete;
  11 import com.google.common.util.concurrent.ListenableFuture;
  12 import com.google.common.util.concurrent.SettableFuture;
  13 import java.util.Arrays;
  14 import java.util.List;
  15 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  16 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  17 import org.slf4j.Logger;
  18 import org.slf4j.LoggerFactory;
  19 import scala.concurrent.Future;
  20
  21 /**
  22  * A cohort proxy implementation for a single-shard transaction commit. If the transaction was a direct commit
  23  * to the shard, this implementation elides the CanCommitTransaction and CommitTransaction messages to the
  24  * shard as an optimization.
  25  *
  26  * @author Thomas Pantelis
  27  */
  28 class SingleCommitCohortProxy extends AbstractThreePhaseCommitCohort<Object> {
  29     private static final Logger LOG = LoggerFactory.getLogger(SingleCommitCohortProxy.class);
  30
  31     private final ActorContext actorContext;
  32     private final Future<Object> cohortFuture;
  33     private final String transactionId;
  34     private volatile DOMStoreThreePhaseCommitCohort delegateCohort = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
  35     private final OperationCallback.Reference operationCallbackRef;
  36
  37     SingleCommitCohortProxy(ActorContext actorContext, Future<Object> cohortFuture, String transactionId,
  38             OperationCallback.Reference operationCallbackRef) {
  39         this.actorContext = actorContext;
  40         this.cohortFuture = cohortFuture;
  41         this.transactionId = transactionId;
  42         this.operationCallbackRef = operationCallbackRef;
  43     }
  44
  45     @Override
  46     public ListenableFuture<Boolean> canCommit() {
  47         LOG.debug("Tx {} canCommit", transactionId);
  48
  49         final SettableFuture<Boolean> returnFuture = SettableFuture.create();
  50
  51         cohortFuture.onComplete(new OnComplete<Object>() {
  52             @Override
  53             public void onComplete(Throwable failure, Object cohortResponse) {
  54                 if(failure != null) {
  55                     operationCallbackRef.get().failure();
  56                     returnFuture.setException(failure);
  57                     return;
  58                 }
  59
  60                 operationCallbackRef.get().success();
  61
  62                 LOG.debug("Tx {} successfully completed direct commit", transactionId);
  63
  64                 // The Future was the result of a direct commit to the shard, essentially eliding the
  65                 // front-end 3PC coordination. We don't really care about the specific Future
  66                 // response object, only that it completed successfully. At this point the Tx is complete
  67                 // so return true. The subsequent preCommit and commit phases will be no-ops, ie return
  68                 // immediate success, to complete the 3PC for the front-end.
  69                 returnFuture.set(Boolean.TRUE);
  70             }
  71         }, actorContext.getClientDispatcher());
  72
  73         return returnFuture;
  74     }
  75
  76     @Override
  77     public ListenableFuture<Void> preCommit() {
  78         return delegateCohort.preCommit();
  79     }
  80
  81     @Override
  82     public ListenableFuture<Void> abort() {
  83         return delegateCohort.abort();
  84     }
  85
  86     @Override
  87     public ListenableFuture<Void> commit() {
  88         return delegateCohort.commit();
  89     }
  90
  91     @Override
  92     List<Future<Object>> getCohortFutures() {
  93         return Arrays.asList(cohortFuture);
  94     }
  95 }