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

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9
  10 package org.opendaylight.controller.cluster.datastore;
  11
  12 import akka.actor.ActorRef;
  13 import akka.actor.PoisonPill;
  14 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  15 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  16 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
  17 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
  18 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  19 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  20 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  21
  22 /**
  23  * @author: syedbahm
  24  * Date: 8/6/14
  25  */
  26 public class ShardWriteTransaction extends ShardTransaction {
  27
  28     private int totalBatchedModificationsReceived;
  29     private Exception lastBatchedModificationsException;
  30     private final ReadWriteShardDataTreeTransaction transaction;
  31
  32     public ShardWriteTransaction(ReadWriteShardDataTreeTransaction transaction, ActorRef shardActor,
  33             ShardStats shardStats, String transactionID, short clientTxVersion) {
  34         super(shardActor, shardStats, transactionID, clientTxVersion);
  35         this.transaction = transaction;
  36     }
  37
  38     @Override
  39     protected ReadWriteShardDataTreeTransaction getDOMStoreTransaction() {
  40         return transaction;
  41     }
  42
  43     @Override
  44     public void handleReceive(Object message) throws Exception {
  45
  46         if (message instanceof BatchedModifications) {
  47             batchedModifications((BatchedModifications)message);
  48         } else {
  49             super.handleReceive(message);
  50         }
  51     }
  52
  53     private void batchedModifications(BatchedModifications batched) {
  54         if (checkClosed()) {
  55             if (batched.isReady()) {
  56                 getSelf().tell(PoisonPill.getInstance(), getSelf());
  57             }
  58             return;
  59         }
  60
  61         try {
  62             for(Modification modification: batched.getModifications()) {
  63                 modification.apply(transaction.getSnapshot());
  64             }
  65
  66             totalBatchedModificationsReceived++;
  67             if(batched.isReady()) {
  68                 if(lastBatchedModificationsException != null) {
  69                     throw lastBatchedModificationsException;
  70                 }
  71
  72                 if(totalBatchedModificationsReceived != batched.getTotalMessagesSent()) {
  73                     throw new IllegalStateException(String.format(
  74                             "The total number of batched messages received %d does not match the number sent %d",
  75                             totalBatchedModificationsReceived, batched.getTotalMessagesSent()));
  76                 }
  77
  78                 readyTransaction(false, batched.isDoCommitOnReady());
  79             } else {
  80                 getSender().tell(new BatchedModificationsReply(batched.getModifications().size()), getSelf());
  81             }
  82         } catch (Exception e) {
  83             lastBatchedModificationsException = e;
  84             getSender().tell(new akka.actor.Status.Failure(e), getSelf());
  85
  86             if(batched.isReady()) {
  87                 getSelf().tell(PoisonPill.getInstance(), getSelf());
  88             }
  89         }
  90     }
  91
  92     protected final void dataExists(DataExists message, final boolean returnSerialized) {
  93         super.dataExists(transaction, message, returnSerialized);
  94     }
  95
  96     protected final void readData(ReadData message, final boolean returnSerialized) {
  97         super.readData(transaction, message, returnSerialized);
  98     }
  99
 100     private boolean checkClosed() {
 101         if (transaction.isClosed()) {
 102             getSender().tell(new akka.actor.Status.Failure(new IllegalStateException("Transaction is closed, no modifications allowed")), getSelf());
 103             return true;
 104         } else {
 105             return false;
 106         }
 107     }
 108
 109     private void readyTransaction(boolean returnSerialized, boolean doImmediateCommit) {
 110         String transactionID = getTransactionID();
 111
 112         LOG.debug("readyTransaction : {}", transactionID);
 113
 114         getShardActor().forward(new ForwardedReadyTransaction(transactionID, getClientTxVersion(),
 115                 transaction, returnSerialized, doImmediateCommit), getContext());
 116
 117         // The shard will handle the commit from here so we're no longer needed - self-destruct.
 118         getSelf().tell(PoisonPill.getInstance(), getSelf());
 119     }
 120 }