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.exceptions.UnknownMessageException;
  19 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  20 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  21 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
  22 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  23 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
  24 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  25 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  26 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  27 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  28 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  29 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
  30 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
  31 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  32 import org.opendaylight.controller.cluster.datastore.modification.ImmutableCompositeModification;
  33 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  34 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  35 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  36 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  37 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  38 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  39 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  40 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  41 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  42 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  43 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  44 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  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.shardActor = shardActor;
  94     this.schemaContext = schemaContext;
  95   }
  96
  97
  98
  99   public static Props props(final DOMStoreReadTransaction transaction,
 100                             final ActorRef shardActor, final SchemaContext schemaContext) {
 101     return Props.create(new Creator<ShardTransaction>() {
 102
 103       @Override
 104       public ShardTransaction create() throws Exception {
 105         return new ShardReadTransaction(transaction, shardActor, schemaContext);
 106       }
 107     });
 108   }
 109
 110   public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadTransaction transaction,
 111                             final ActorRef shardActor, final SchemaContext schemaContext) {
 112     return Props.create(new Creator<ShardTransaction>() {
 113
 114       @Override
 115       public ShardTransaction create() throws Exception {
 116         return new ShardReadTransaction(transactionChain, transaction, shardActor, schemaContext);
 117       }
 118     });
 119   }
 120
 121   public static Props props(final DOMStoreReadWriteTransaction transaction,
 122                             final ActorRef shardActor, final SchemaContext schemaContext) {
 123     return Props.create(new Creator<ShardTransaction>() {
 124
 125       @Override
 126       public ShardTransaction create() throws Exception {
 127         return new ShardReadWriteTransaction(transaction, shardActor, schemaContext);
 128       }
 129     });
 130   }
 131
 132   public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
 133                             final ActorRef shardActor, final SchemaContext schemaContext) {
 134     return Props.create(new Creator<ShardTransaction>() {
 135
 136       @Override
 137       public ShardTransaction create() throws Exception {
 138         return new ShardReadWriteTransaction(transactionChain, transaction, shardActor, schemaContext);
 139       }
 140     });
 141   }
 142
 143
 144   public static Props props(final DOMStoreWriteTransaction transaction,
 145                             final ActorRef shardActor, final SchemaContext schemaContext) {
 146     return Props.create(new Creator<ShardTransaction>() {
 147
 148       @Override
 149       public ShardTransaction create() throws Exception {
 150         return new ShardWriteTransaction(transaction, shardActor, schemaContext);
 151       }
 152     });
 153   }
 154
 155   public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreWriteTransaction transaction,
 156                             final ActorRef shardActor, final SchemaContext schemaContext) {
 157     return Props.create(new Creator<ShardTransaction>() {
 158
 159       @Override
 160       public ShardTransaction create() throws Exception {
 161         return new ShardWriteTransaction(transactionChain, transaction, shardActor, schemaContext);
 162       }
 163     });
 164   }
 165
 166
 167   @Override
 168   public void handleReceive(Object message) throws Exception {
 169      if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) {
 170       closeTransaction(new CloseTransaction());
 171     } else if (message instanceof GetCompositedModification) {
 172       // This is here for testing only
 173       getSender().tell(new GetCompositeModificationReply(
 174           new ImmutableCompositeModification(modification)), getSelf());
 175     }else{
 176          throw new UnknownMessageException(message);
 177     }
 178   }
 179
 180   abstract protected  void closeTransaction(CloseTransaction message);
 181
 182   protected void readData(DOMStoreReadTransaction transaction,ReadData message) {
 183     final ActorRef sender = getSender();
 184     final ActorRef self = getSelf();
 185     final YangInstanceIdentifier path = message.getPath();
 186     final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
 187         transaction.read(path);
 188
 189     future.addListener(new Runnable() {
 190       @Override
 191       public void run() {
 192         try {
 193           Optional<NormalizedNode<?, ?>> optional = future.get();
 194           if (optional.isPresent()) {
 195             sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
 196           } else {
 197             sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self);
 198           }
 199         } catch (Exception e) {
 200             sender.tell(new akka.actor.Status.Failure(new ReadFailedException( "An Exception occurred  when reading data from path : "
 201                 + path.toString(),e)),self);
 202         }
 203
 204       }
 205     }, getContext().dispatcher());
 206   }
 207
 208
 209   protected void writeData(DOMStoreWriteTransaction transaction, WriteData message) {
 210     modification.addModification(
 211         new WriteModification(message.getPath(), message.getData(),schemaContext));
 212     LOG.debug("writeData at path : " + message.getPath().toString());
 213
 214     try {
 215         transaction.write(message.getPath(), message.getData());
 216         getSender().tell(new WriteDataReply().toSerializable(), getSelf());
 217     }catch(Exception e){
 218         getSender().tell(new akka.actor.Status.Failure(e), getSelf());
 219     }
 220   }
 221
 222   protected void mergeData(DOMStoreWriteTransaction transaction, MergeData message) {
 223     modification.addModification(
 224         new MergeModification(message.getPath(), message.getData(), schemaContext));
 225     LOG.debug("mergeData at path : " + message.getPath().toString());
 226     try {
 227         transaction.merge(message.getPath(), message.getData());
 228         getSender().tell(new MergeDataReply().toSerializable(), getSelf());
 229     }catch(Exception e){
 230         getSender().tell(new akka.actor.Status.Failure(e), getSelf());
 231     }
 232   }
 233
 234   protected void deleteData(DOMStoreWriteTransaction transaction, DeleteData message) {
 235     LOG.debug("deleteData at path : " + message.getPath().toString());
 236     modification.addModification(new DeleteModification(message.getPath()));
 237     try {
 238         transaction.delete(message.getPath());
 239         getSender().tell(new DeleteDataReply().toSerializable(), getSelf());
 240     }catch(Exception e){
 241         getSender().tell(new akka.actor.Status.Failure(e), getSelf());
 242     }
 243   }
 244
 245   protected void readyTransaction(DOMStoreWriteTransaction transaction, ReadyTransaction message) {
 246     DOMStoreThreePhaseCommitCohort cohort =  transaction.ready();
 247     ActorRef cohortActor = getContext().actorOf(
 248         ThreePhaseCommitCohort.props(cohort, shardActor, modification), "cohort");
 249     getSender()
 250         .tell(new ReadyTransactionReply(cohortActor.path()).toSerializable(), getSelf());
 251
 252   }
 253
 254
 255   // These classes are in here for test purposes only
 256
 257
 258   static class GetCompositedModification {
 259
 260   }
 261
 262
 263   static class GetCompositeModificationReply {
 264     private final CompositeModification modification;
 265
 266
 267     GetCompositeModificationReply(CompositeModification modification) {
 268       this.modification = modification;
 269     }
 270
 271
 272     public CompositeModification getModification() {
 273       return modification;
 274     }
 275   }
 276 }