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 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.PoisonPill;
  13 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  14 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  15 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
  16 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
  17 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  18 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  19 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  20
  21 /**
  22  * Actor for a shard write-only transaction.
  23  *
  24  * @author syedbahm
  25  */
  26 @Deprecated(since = "9.0.0", forRemoval = true)
  27 public class ShardWriteTransaction extends ShardTransaction {
  28     private int totalBatchedModificationsReceived;
  29     private Exception lastBatchedModificationsException;
  30     private final ReadWriteShardDataTreeTransaction transaction;
  31
  32     public ShardWriteTransaction(final ReadWriteShardDataTreeTransaction transaction, final ActorRef shardActor,
  33             final ShardStats shardStats) {
  34         super(shardActor, shardStats, transaction.getIdentifier());
  35         this.transaction = transaction;
  36     }
  37
  38     @Override
  39     protected ReadWriteShardDataTreeTransaction getDOMStoreTransaction() {
  40         return transaction;
  41     }
  42
  43     @Override
  44     public void handleReceive(final Object message) {
  45         if (message instanceof BatchedModifications) {
  46             batchedModifications((BatchedModifications)message);
  47         } else {
  48             super.handleReceive(message);
  49         }
  50     }
  51
  52     @SuppressWarnings("checkstyle:IllegalCatch")
  53     private void batchedModifications(final 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(batched);
  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(final DataExists message) {
  93         super.dataExists(transaction, message);
  94     }
  95
  96     protected final void readData(final ReadData message) {
  97         super.readData(transaction, message);
  98     }
  99
 100     private boolean checkClosed() {
 101         final boolean ret = transaction.isClosed();
 102         if (ret) {
 103             getSender().tell(new akka.actor.Status.Failure(new IllegalStateException(
 104                     "Transaction is closed, no modifications allowed")), getSelf());
 105         }
 106         return ret;
 107     }
 108
 109     private void readyTransaction(final BatchedModifications batched) {
 110         TransactionIdentifier transactionID = getTransactionId();
 111
 112         LOG.debug("readyTransaction : {}", transactionID);
 113
 114         getShardActor().forward(new ForwardedReadyTransaction(transactionID, batched.getVersion(),
 115                 transaction, batched.isDoCommitOnReady(), batched.getParticipatingShardNames()), 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 }