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     protected void onTransactionReady(List<Future<ActorSelection>> readyFutures) {
  48         LOG.debug("onTransactionReady {} pending readyFutures size {} chain {}", getIdentifier(),
  49                 readyFutures.size(), getTransactionChainId());
  50         this.readyFutures = readyFutures;
  51     }
  52
  53     /**
  54      * This method is overridden to ensure the previous Tx's ready operations complete
  55      * before we initiate the next Tx in the chain to avoid creation failures if the
  56      * previous Tx's ready operations haven't completed yet.
  57      */
  58     @Override
  59     protected Future<ActorSelection> sendFindPrimaryShardAsync(final String shardName) {
  60         // Check if there are any previous ready Futures, otherwise let the super class handle it.
  61         if(previousReadyFutures.isEmpty()) {
  62             return super.sendFindPrimaryShardAsync(shardName);
  63         }
  64
  65         if (LOG.isDebugEnabled()) {
  66             LOG.debug("Waiting for {} previous ready futures for Tx {} on chain {}",
  67                     previousReadyFutures.size(), getIdentifier(), getTransactionChainId());
  68         }
  69
  70         // Combine the ready Futures into 1.
  71         Future<Iterable<ActorSelection>> combinedFutures = akka.dispatch.Futures.sequence(
  72                 previousReadyFutures, getActorContext().getClientDispatcher());
  73
  74         // Add a callback for completion of the combined Futures.
  75         final Promise<ActorSelection> returnPromise = akka.dispatch.Futures.promise();
  76         OnComplete<Iterable<ActorSelection>> onComplete = new OnComplete<Iterable<ActorSelection>>() {
  77             @Override
  78             public void onComplete(Throwable failure, Iterable<ActorSelection> notUsed) {
  79                 if(failure != null) {
  80                     // A Ready Future failed so fail the returned Promise.
  81                     returnPromise.failure(failure);
  82                 } else {
  83                     LOG.debug("Previous Tx readied - sending FindPrimaryShard for {} on chain {}",
  84                             getIdentifier(), getTransactionChainId());
  85
  86                     // Send the FindPrimaryShard message and use the resulting Future to complete the
  87                     // returned Promise.
  88                     returnPromise.completeWith(ChainedTransactionProxy.super.sendFindPrimaryShardAsync(shardName));
  89                 }
  90             }
  91         };
  92
  93         combinedFutures.onComplete(onComplete, getActorContext().getClientDispatcher());
  94
  95         return returnPromise.future();
  96     }
  97 }