opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.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.ActorRef;
  12 import akka.actor.Props;
  13 import akka.event.Logging;
  14 import akka.event.LoggingAdapter;
  15 import akka.japi.Creator;
  16 import com.google.common.base.Optional;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  19 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  20 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
  21 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  22 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
  23 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  24 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  25 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  26 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  27 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  28 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
  29 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
  30 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  31 import org.opendaylight.controller.cluster.datastore.modification.ImmutableCompositeModification;
  32 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  33 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  34 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  35 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  36 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  37 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  38 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  39 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  40 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  42 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  43
  44 import java.util.concurrent.ExecutionException;
  45
  46 /**
  47  * The ShardTransaction Actor represents a remote transaction
  48  * <p>
  49  * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
  50  * </p>
  51  * <p>
  52  * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
  53  * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
  54  * time there are no known advantages for creating a read-only or write-only transaction which may change over time
  55  * at which point we can optimize things in the distributed store as well.
  56  * </p>
  57  * <p>
  58  * Handles Messages <br/>
  59  * ---------------- <br/>
  60  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
  61  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.WriteData}
  62  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.MergeData}
  63  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.DeleteData}
  64  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction}
  65  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
  66  * </p>
  67  */
  68 public abstract class ShardTransaction extends AbstractUntypedActor {
  69
  70   private final ActorRef shardActor;
  71   protected final SchemaContext schemaContext;
  72
  73   // FIXME : see below
  74   // If transactionChain is not null then this transaction is part of a
  75   // transactionChain. Not really clear as to what that buys us
  76   private final DOMStoreTransactionChain transactionChain;
  77
  78
  79   private final MutableCompositeModification modification =
  80       new MutableCompositeModification();
  81
  82   private final LoggingAdapter log =
  83       Logging.getLogger(getContext().system(), this);
  84
  85   protected ShardTransaction(
  86                           ActorRef shardActor, SchemaContext schemaContext) {
  87     this(null,  shardActor, schemaContext);
  88   }
  89
  90   protected ShardTransaction(DOMStoreTransactionChain transactionChain,
  91                           ActorRef shardActor, SchemaContext schemaContext) {
  92     this.transactionChain = transactionChain;
  93     //this.transaction = transaction;
  94     this.shardActor = shardActor;
  95     this.schemaContext = schemaContext;
  96   }
  97
  98
  99
 100   public static Props props(final DOMStoreReadTransaction transaction,
 101                             final ActorRef shardActor, final SchemaContext schemaContext) {
 102     return Props.create(new Creator<ShardTransaction>() {
 103
 104       @Override
 105       public ShardTransaction create() throws Exception {
 106         return new ShardReadTransaction(transaction, shardActor, schemaContext);
 107       }
 108     });
 109   }
 110
 111   public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadTransaction transaction,
 112                             final ActorRef shardActor, final SchemaContext schemaContext) {
 113     return Props.create(new Creator<ShardTransaction>() {
 114
 115       @Override
 116       public ShardTransaction create() throws Exception {
 117         return new ShardReadTransaction(transactionChain, transaction, shardActor, schemaContext);
 118       }
 119     });
 120   }
 121
 122   public static Props props(final DOMStoreReadWriteTransaction transaction,
 123                             final ActorRef shardActor, final SchemaContext schemaContext) {
 124     return Props.create(new Creator<ShardTransaction>() {
 125
 126       @Override
 127       public ShardTransaction create() throws Exception {
 128         return new ShardReadWriteTransaction(transaction, shardActor, schemaContext);
 129       }
 130     });
 131   }
 132
 133   public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
 134                             final ActorRef shardActor, final SchemaContext schemaContext) {
 135     return Props.create(new Creator<ShardTransaction>() {
 136
 137       @Override
 138       public ShardTransaction create() throws Exception {
 139         return new ShardReadWriteTransaction(transactionChain, transaction, shardActor, schemaContext);
 140       }
 141     });
 142   }
 143
 144
 145   public static Props props(final DOMStoreWriteTransaction transaction,
 146                             final ActorRef shardActor, final SchemaContext schemaContext) {
 147     return Props.create(new Creator<ShardTransaction>() {
 148
 149       @Override
 150       public ShardTransaction create() throws Exception {
 151         return new ShardWriteTransaction(transaction, shardActor, schemaContext);
 152       }
 153     });
 154   }
 155
 156   public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreWriteTransaction transaction,
 157                             final ActorRef shardActor, final SchemaContext schemaContext) {
 158     return Props.create(new Creator<ShardTransaction>() {
 159
 160       @Override
 161       public ShardTransaction create() throws Exception {
 162         return new ShardWriteTransaction(transactionChain, transaction, shardActor, schemaContext);
 163       }
 164     });
 165   }
 166
 167
 168   @Override
 169   public void handleReceive(Object message) throws Exception {
 170      if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) {
 171       closeTransaction(new CloseTransaction());
 172     } else if (message instanceof GetCompositedModification) {
 173       // This is here for testing only
 174       getSender().tell(new GetCompositeModificationReply(
 175           new ImmutableCompositeModification(modification)), getSelf());
 176     }else{
 177       throw new Exception ("ShardTransaction:handleRecieve received an unknown message"+message);
 178     }
 179   }
 180
 181   abstract protected  void closeTransaction(CloseTransaction message);
 182
 183   protected void readData(DOMStoreReadTransaction transaction,ReadData message) {
 184     final ActorRef sender = getSender();
 185     final ActorRef self = getSelf();
 186     final YangInstanceIdentifier path = message.getPath();
 187     final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
 188         transaction.read(path);
 189
 190     future.addListener(new Runnable() {
 191       @Override
 192       public void run() {
 193         try {
 194           Optional<NormalizedNode<?, ?>> optional = future.get();
 195           if (optional.isPresent()) {
 196             sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
 197           } else {
 198             sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self);
 199           }
 200         } catch (InterruptedException | ExecutionException e) {
 201           log.error(e,
 202               "An exception happened when reading data from path : "
 203                   + path.toString());
 204         }
 205
 206       }
 207     }, getContext().dispatcher());
 208   }
 209
 210
 211   protected void writeData(DOMStoreWriteTransaction transaction, WriteData message) {
 212     modification.addModification(
 213         new WriteModification(message.getPath(), message.getData(),schemaContext));
 214     LOG.debug("writeData at path : " + message.getPath().toString());
 215     transaction.write(message.getPath(), message.getData());
 216     getSender().tell(new WriteDataReply().toSerializable(), getSelf());
 217   }
 218
 219   protected void mergeData(DOMStoreWriteTransaction transaction, MergeData message) {
 220     modification.addModification(
 221         new MergeModification(message.getPath(), message.getData(), schemaContext));
 222     LOG.debug("mergeData at path : " + message.getPath().toString());
 223     transaction.merge(message.getPath(), message.getData());
 224     getSender().tell(new MergeDataReply().toSerializable(), getSelf());
 225   }
 226
 227   protected void deleteData(DOMStoreWriteTransaction transaction, DeleteData message) {
 228     modification.addModification(new DeleteModification(message.getPath()));
 229     transaction.delete(message.getPath());
 230     getSender().tell(new DeleteDataReply().toSerializable(), getSelf());
 231   }
 232
 233   protected void readyTransaction(DOMStoreWriteTransaction transaction, ReadyTransaction message) {
 234     DOMStoreThreePhaseCommitCohort cohort =  transaction.ready();
 235     ActorRef cohortActor = getContext().actorOf(
 236         ThreePhaseCommitCohort.props(cohort, shardActor, modification), "cohort");
 237     getSender()
 238         .tell(new ReadyTransactionReply(cohortActor.path()).toSerializable(), getSelf());
 239
 240   }
 241
 242
 243   // These classes are in here for test purposes only
 244
 245
 246   static class GetCompositedModification {
 247
 248   }
 249
 250
 251   static class GetCompositeModificationReply {
 252     private final CompositeModification modification;
 253
 254
 255     GetCompositeModificationReply(CompositeModification modification) {
 256       this.modification = modification;
 257     }
 258
 259
 260     public CompositeModification getModification() {
 261       return modification;
 262     }
 263   }
 264 }