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