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 DOMStoreReadWriteTransaction transaction;
  69
  70   private final MutableCompositeModification modification = new MutableCompositeModification();
  71
  72   private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  73
  74   public ShardTransaction(DOMStoreReadWriteTransaction transaction) {
  75     this.transaction = transaction;
  76   }
  77
  78
  79   public static Props props(final DOMStoreReadWriteTransaction transaction){
  80     return Props.create(new Creator<ShardTransaction>(){
  81
  82       @Override
  83       public ShardTransaction create() throws Exception {
  84         return new ShardTransaction(transaction);
  85       }
  86     });
  87   }
  88
  89   @Override
  90   public void onReceive(Object message) throws Exception {
  91     if(message instanceof ReadData){
  92       readData((ReadData) message);
  93     } else if(message instanceof WriteData){
  94       writeData((WriteData) message);
  95     } else if(message instanceof MergeData){
  96       mergeData((MergeData) message);
  97     } else if(message instanceof DeleteData){
  98       deleteData((DeleteData) message);
  99     } else if(message instanceof ReadyTransaction){
 100       readyTransaction((ReadyTransaction) message);
 101     } else if(message instanceof CloseTransaction){
 102       closeTransaction((CloseTransaction) message);
 103     } else if(message instanceof GetCompositedModification){
 104       // This is here for testing only
 105       getSender().tell(new GetCompositeModificationReply(new ImmutableCompositeModification(modification)), getSelf());
 106     }
 107   }
 108
 109   private void readData(ReadData message) {
 110     final ActorRef sender = getSender();
 111     final ActorRef self = getSelf();
 112     final InstanceIdentifier path = message.getPath();
 113     final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = transaction.read(path);
 114
 115     future.addListener(new Runnable() {
 116       @Override
 117       public void run() {
 118         try {
 119           Optional<NormalizedNode<?, ?>> optional = future.get();
 120           if(optional.isPresent()){
 121             sender.tell(new ReadDataReply(optional.get()), self);
 122           } else {
 123             //TODO : Need to decide what to do here
 124           }
 125         } catch (InterruptedException | ExecutionException e) {
 126           log.error(e, "An exception happened when reading data from path : " + path.toString());
 127         }
 128
 129       }
 130     }, getContext().dispatcher());
 131   }
 132
 133
 134   private void writeData(WriteData message){
 135     modification.addModification(new WriteModification(message.getPath(), message.getData()));
 136     transaction.write(message.getPath(), message.getData());
 137     getSender().tell(new WriteDataReply(), getSelf());
 138   }
 139
 140   private void mergeData(MergeData message){
 141     modification.addModification(new MergeModification(message.getPath(), message.getData()));
 142     transaction.merge(message.getPath(), message.getData());
 143     getSender().tell(new MergeDataReply(), getSelf());
 144   }
 145
 146   private void deleteData(DeleteData message){
 147     modification.addModification(new DeleteModification(message.getPath()));
 148     transaction.delete(message.getPath());
 149     getSender().tell(new DeleteDataReply(), getSelf());
 150   }
 151
 152   private void readyTransaction(ReadyTransaction message){
 153     DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
 154     ActorRef cohortActor = getContext().actorOf(ThreePhaseCommitCohort.props(cohort));
 155     getSender().tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
 156
 157   }
 158
 159   private void closeTransaction(CloseTransaction message){
 160     transaction.close();
 161     getSender().tell(new CloseTransactionReply(), getSelf());
 162   }
 163
 164
 165   // These classes are in here for test purposes only
 166
 167   static class GetCompositedModification {
 168
 169   }
 170
 171   static class GetCompositeModificationReply {
 172     private final CompositeModification modification;
 173
 174
 175     GetCompositeModificationReply(CompositeModification modification) {
 176       this.modification = modification;
 177     }
 178
 179
 180     public CompositeModification getModification() {
 181       return modification;
 182     }
 183   }
 184 }