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

   1 /*
   2  * Copyright (c) 2014 Cisco 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
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorSelection;
  12 import com.google.common.base.Preconditions;
  13 import java.util.Collections;
  14 import java.util.List;
  15 import java.util.concurrent.atomic.AtomicInteger;
  16 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
  17 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  18 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  19 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  20 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  21 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  22 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  23 import scala.concurrent.Future;
  24
  25 /**
  26  * TransactionChainProxy acts as a proxy for a DOMStoreTransactionChain created on a remote shard
  27  */
  28 public class TransactionChainProxy implements DOMStoreTransactionChain {
  29
  30     private interface State {
  31         boolean isReady();
  32
  33         List<Future<ActorSelection>> getPreviousReadyFutures();
  34     }
  35
  36     private static class Allocated implements State {
  37         private final ChainedTransactionProxy transaction;
  38
  39         Allocated(ChainedTransactionProxy transaction) {
  40             this.transaction = transaction;
  41         }
  42
  43         @Override
  44         public boolean isReady() {
  45             return transaction.isReady();
  46         }
  47
  48         @Override
  49         public List<Future<ActorSelection>> getPreviousReadyFutures() {
  50             return transaction.getReadyFutures();
  51         }
  52     }
  53
  54     private static abstract class AbstractDefaultState implements State {
  55         @Override
  56         public List<Future<ActorSelection>> getPreviousReadyFutures() {
  57             return Collections.emptyList();
  58         }
  59     }
  60
  61     private static final State IDLE_STATE = new AbstractDefaultState() {
  62         @Override
  63         public boolean isReady() {
  64             return true;
  65         }
  66     };
  67
  68     private static final State CLOSED_STATE = new AbstractDefaultState() {
  69         @Override
  70         public boolean isReady() {
  71             throw new TransactionChainClosedException("Transaction chain has been closed");
  72         }
  73     };
  74
  75     private static final AtomicInteger counter = new AtomicInteger(0);
  76
  77     private final ActorContext actorContext;
  78     private final String transactionChainId;
  79     private volatile State currentState = IDLE_STATE;
  80
  81     public TransactionChainProxy(ActorContext actorContext) {
  82         this.actorContext = actorContext;
  83         transactionChainId = actorContext.getCurrentMemberName() + "-txn-chain-" + counter.incrementAndGet();
  84     }
  85
  86     public String getTransactionChainId() {
  87         return transactionChainId;
  88     }
  89
  90     @Override
  91     public DOMStoreReadTransaction newReadOnlyTransaction() {
  92         State localState = currentState;
  93         checkReadyState(localState);
  94
  95         return new ChainedTransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY,
  96                 transactionChainId, localState.getPreviousReadyFutures());
  97     }
  98
  99     @Override
 100     public DOMStoreReadWriteTransaction newReadWriteTransaction() {
 101         actorContext.acquireTxCreationPermit();
 102         return allocateWriteTransaction(TransactionProxy.TransactionType.READ_WRITE);
 103     }
 104
 105     @Override
 106     public DOMStoreWriteTransaction newWriteOnlyTransaction() {
 107         actorContext.acquireTxCreationPermit();
 108         return allocateWriteTransaction(TransactionProxy.TransactionType.WRITE_ONLY);
 109     }
 110
 111     @Override
 112     public void close() {
 113         currentState = CLOSED_STATE;
 114
 115         // Send a close transaction chain request to each and every shard
 116         actorContext.broadcast(new CloseTransactionChain(transactionChainId));
 117     }
 118
 119     private ChainedTransactionProxy allocateWriteTransaction(TransactionProxy.TransactionType type) {
 120         State localState = currentState;
 121
 122         checkReadyState(localState);
 123
 124         // Pass the ready Futures from the previous Tx.
 125         ChainedTransactionProxy txProxy = new ChainedTransactionProxy(actorContext, type,
 126                 transactionChainId, localState.getPreviousReadyFutures());
 127
 128         currentState = new Allocated(txProxy);
 129
 130         return txProxy;
 131     }
 132
 133     private void checkReadyState(State state) {
 134         Preconditions.checkState(state.isReady(), "Previous transaction is not ready yet");
 135     }
 136 }