2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore;
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;
19 final class ChainedTransactionProxy extends TransactionProxy {
20 private static final Logger LOG = LoggerFactory.getLogger(ChainedTransactionProxy.class);
23 * Stores the ready Futures from the previous Tx in the chain.
25 private final List<Future<ActorSelection>> previousReadyFutures;
28 * Stores the ready Futures from this transaction when it is readied.
30 private volatile List<Future<ActorSelection>> readyFutures;
32 ChainedTransactionProxy(ActorContext actorContext, TransactionType transactionType,
33 String transactionChainId, List<Future<ActorSelection>> previousReadyFutures) {
34 super(actorContext, transactionType, transactionChainId);
35 this.previousReadyFutures = previousReadyFutures;
38 List<Future<ActorSelection>> getReadyFutures() {
43 return readyFutures != null;
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());
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.
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);
67 if (LOG.isDebugEnabled()) {
68 LOG.debug("Waiting for {} previous ready futures for Tx {} on chain {}",
69 previousReadyFutures.size(), getIdentifier(), getTransactionChainId());
72 // Combine the ready Futures into 1.
73 Future<Iterable<ActorSelection>> combinedFutures = akka.dispatch.Futures.sequence(
74 previousReadyFutures, getActorContext().getClientDispatcher());
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>>() {
80 public void onComplete(Throwable failure, Iterable<ActorSelection> notUsed) {
82 // A Ready Future failed so fail the returned Promise.
83 returnPromise.failure(failure);
85 LOG.debug("Previous Tx readied - sending FindPrimaryShard for {} on chain {}",
86 getIdentifier(), getTransactionChainId());
88 // Send the FindPrimaryShard message and use the resulting Future to complete the
90 returnPromise.completeWith(ChainedTransactionProxy.super.sendFindPrimaryShardAsync(shardName));
95 combinedFutures.onComplete(onComplete, getActorContext().getClientDispatcher());
97 return returnPromise.future();