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 protected void onTransactionReady(List<Future<ActorSelection>> readyFutures) {
48 LOG.debug("onTransactionReady {} pending readyFutures size {} chain {}", getIdentifier(),
49 readyFutures.size(), getTransactionChainId());
50 this.readyFutures = readyFutures;
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.
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);
65 if (LOG.isDebugEnabled()) {
66 LOG.debug("Waiting for {} previous ready futures for Tx {} on chain {}",
67 previousReadyFutures.size(), getIdentifier(), getTransactionChainId());
70 // Combine the ready Futures into 1.
71 Future<Iterable<ActorSelection>> combinedFutures = akka.dispatch.Futures.sequence(
72 previousReadyFutures, getActorContext().getClientDispatcher());
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>>() {
78 public void onComplete(Throwable failure, Iterable<ActorSelection> notUsed) {
80 // A Ready Future failed so fail the returned Promise.
81 returnPromise.failure(failure);
83 LOG.debug("Previous Tx readied - sending FindPrimaryShard for {} on chain {}",
84 getIdentifier(), getTransactionChainId());
86 // Send the FindPrimaryShard message and use the resulting Future to complete the
88 returnPromise.completeWith(ChainedTransactionProxy.super.sendFindPrimaryShardAsync(shardName));
93 combinedFutures.onComplete(onComplete, getActorContext().getClientDispatcher());
95 return returnPromise.future();