opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ChainedTransactionProxy.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.actor.ActorSelection;
  11 import akka.dispatch.OnComplete;
  12 import java.util.List;
  13 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  14 import org.slf4j.Logger;
  15 import org.slf4j.LoggerFactory;
  16 import scala.concurrent.Future;
  17 import scala.concurrent.Promise;
  18
  19 final class ChainedTransactionProxy extends TransactionProxy {
  20     private static final Logger LOG = LoggerFactory.getLogger(ChainedTransactionProxy.class);
  21
  22     /**
  23      * Stores the ready Futures from the previous Tx in the chain.
  24      */
  25     private final List<Future<ActorSelection>> previousReadyFutures;
  26
  27     /**
  28      * Stores the ready Futures from this transaction when it is readied.
  29      */
  30     private volatile List<Future<ActorSelection>> readyFutures;
  31
  32     ChainedTransactionProxy(ActorContext actorContext, TransactionType transactionType,
  33             String transactionChainId, List<Future<ActorSelection>> previousReadyFutures) {
  34         super(actorContext, transactionType, transactionChainId);
  35         this.previousReadyFutures = previousReadyFutures;
  36     }
  37
  38     List<Future<ActorSelection>> getReadyFutures() {
  39         return readyFutures;
  40     }
  41
  42     boolean isReady() {
  43         return readyFutures != null;
  44     }
  45
  46     @Override
  47     public AbstractThreePhaseCommitCohort ready() {
  48         final AbstractThreePhaseCommitCohort ret = super.ready();
  49         readyFutures = ret.getCohortFutures();
  50         LOG.debug("onTransactionReady {} pending readyFutures size {} chain {}", getIdentifier(),
  51             readyFutures.size(), getTransactionChainId());
  52         return ret;
  53     }
  54
  55     /**
  56      * This method is overridden to ensure the previous Tx's ready operations complete
  57      * before we initiate the next Tx in the chain to avoid creation failures if the
  58      * previous Tx's ready operations haven't completed yet.
  59      */
  60     @Override
  61     protected Future<ActorSelection> sendFindPrimaryShardAsync(final String shardName) {
  62         // Check if there are any previous ready Futures, otherwise let the super class handle it.
  63         if(previousReadyFutures.isEmpty()) {
  64             return super.sendFindPrimaryShardAsync(shardName);
  65         }
  66
  67         if (LOG.isDebugEnabled()) {
  68             LOG.debug("Waiting for {} previous ready futures for Tx {} on chain {}",
  69                     previousReadyFutures.size(), getIdentifier(), getTransactionChainId());
  70         }
  71
  72         // Combine the ready Futures into 1.
  73         Future<Iterable<ActorSelection>> combinedFutures = akka.dispatch.Futures.sequence(
  74                 previousReadyFutures, getActorContext().getClientDispatcher());
  75
  76         // Add a callback for completion of the combined Futures.
  77         final Promise<ActorSelection> returnPromise = akka.dispatch.Futures.promise();
  78         OnComplete<Iterable<ActorSelection>> onComplete = new OnComplete<Iterable<ActorSelection>>() {
  79             @Override
  80             public void onComplete(Throwable failure, Iterable<ActorSelection> notUsed) {
  81                 if(failure != null) {
  82                     // A Ready Future failed so fail the returned Promise.
  83                     returnPromise.failure(failure);
  84                 } else {
  85                     LOG.debug("Previous Tx readied - sending FindPrimaryShard for {} on chain {}",
  86                             getIdentifier(), getTransactionChainId());
  87
  88                     // Send the FindPrimaryShard message and use the resulting Future to complete the
  89                     // returned Promise.
  90                     returnPromise.completeWith(ChainedTransactionProxy.super.sendFindPrimaryShardAsync(shardName));
  91                 }
  92             }
  93         };
  94
  95         combinedFutures.onComplete(onComplete, getActorContext().getClientDispatcher());
  96
  97         return returnPromise.future();
  98     }
  99 }