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.actor.UntypedActor;
  14 import akka.event.Logging;
  15 import akka.event.LoggingAdapter;
  16 import akka.japi.Creator;
  17 import com.google.common.base.Optional;
  18 import com.google.common.util.concurrent.ListenableFuture;
  19 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  20 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
  21 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  22 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
  23 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  24 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
  25 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  26 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  27 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  28 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  29 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  30 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
  31 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
  32 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  33 import org.opendaylight.controller.cluster.datastore.modification.ImmutableCompositeModification;
  34 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  35 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  36 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  37 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  38 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  39 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  40 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  41
  42 import java.util.concurrent.ExecutionException;
  43
  44 /**
  45  * The ShardTransaction Actor represents a remote transaction
  46  *<p>
  47  * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
  48  *</p>
  49  *<p>
  50  * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
  51  * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
  52  * time there are no known advantages for creating a read-only or write-only transaction which may change over time
  53  * at which point we can optimize things in the distributed store as well.
  54  *</p>
  55  *<p>
  56  * Handles Messages <br/>
  57  * ---------------- <br/>
  58  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
  59  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.WriteData}
  60  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.MergeData}
  61  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.DeleteData}
  62  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction}
  63  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
  64  * </p>
  65  */
  66 public class ShardTransaction extends UntypedActor {
  67
  68   private final ActorRef shardActor;
  69
  70   private final DOMStoreReadWriteTransaction transaction;
  71
  72   private final MutableCompositeModification modification = new MutableCompositeModification();
  73
  74   private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  75
  76   public ShardTransaction(DOMStoreReadWriteTransaction transaction, ActorRef shardActor) {
  77     this.transaction = transaction;
  78     this.shardActor = shardActor;
  79   }
  80
  81
  82   public static Props props(final DOMStoreReadWriteTransaction transaction, final ActorRef shardActor){
  83     return Props.create(new Creator<ShardTransaction>(){
  84
  85       @Override
  86       public ShardTransaction create() throws Exception {
  87         return new ShardTransaction(transaction, shardActor);
  88       }
  89     });
  90   }
  91
  92   @Override
  93   public void onReceive(Object message) throws Exception {
  94     if(message instanceof ReadData){
  95       readData((ReadData) message);
  96     } else if(message instanceof WriteData){
  97       writeData((WriteData) message);
  98     } else if(message instanceof MergeData){
  99       mergeData((MergeData) message);
 100     } else if(message instanceof DeleteData){
 101       deleteData((DeleteData) message);
 102     } else if(message instanceof ReadyTransaction){
 103       readyTransaction((ReadyTransaction) message);
 104     } else if(message instanceof CloseTransaction){
 105       closeTransaction((CloseTransaction) message);
 106     } else if(message instanceof GetCompositedModification){
 107       // This is here for testing only
 108       getSender().tell(new GetCompositeModificationReply(new ImmutableCompositeModification(modification)), getSelf());
 109     }
 110   }
 111
 112   private void readData(ReadData message) {
 113     final ActorRef sender = getSender();
 114     final ActorRef self = getSelf();
 115     final InstanceIdentifier path = message.getPath();
 116     final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = transaction.read(path);
 117
 118     future.addListener(new Runnable() {
 119       @Override
 120       public void run() {
 121         try {
 122           Optional<NormalizedNode<?, ?>> optional = future.get();
 123           if(optional.isPresent()){
 124             sender.tell(new ReadDataReply(optional.get()), self);
 125           } else {
 126             //TODO : Need to decide what to do here
 127           }
 128         } catch (InterruptedException | ExecutionException e) {
 129           log.error(e, "An exception happened when reading data from path : " + path.toString());
 130         }
 131
 132       }
 133     }, getContext().dispatcher());
 134   }
 135
 136
 137   private void writeData(WriteData message){
 138     modification.addModification(new WriteModification(message.getPath(), message.getData()));
 139     transaction.write(message.getPath(), message.getData());
 140     getSender().tell(new WriteDataReply(), getSelf());
 141   }
 142
 143   private void mergeData(MergeData message){
 144     modification.addModification(new MergeModification(message.getPath(), message.getData()));
 145     transaction.merge(message.getPath(), message.getData());
 146     getSender().tell(new MergeDataReply(), getSelf());
 147   }
 148
 149   private void deleteData(DeleteData message){
 150     modification.addModification(new DeleteModification(message.getPath()));
 151     transaction.delete(message.getPath());
 152     getSender().tell(new DeleteDataReply(), getSelf());
 153   }
 154
 155   private void readyTransaction(ReadyTransaction message){
 156     DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
 157     ActorRef cohortActor = getContext().actorOf(ThreePhaseCommitCohort.props(cohort, shardActor, modification));
 158     getSender().tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
 159
 160   }
 161
 162   private void closeTransaction(CloseTransaction message){
 163     transaction.close();
 164     getSender().tell(new CloseTransactionReply(), getSelf());
 165   }
 166
 167
 168   // These classes are in here for test purposes only
 169
 170   static class GetCompositedModification {
 171
 172   }
 173
 174   static class GetCompositeModificationReply {
 175     private final CompositeModification modification;
 176
 177
 178     GetCompositeModificationReply(CompositeModification modification) {
 179       this.modification = modification;
 180     }
 181
 182
 183     public CompositeModification getModification() {
 184       return modification;
 185     }
 186   }
 187 }