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.sal.core.spi.data.DOMStoreReadWriteTransaction;
  32 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  33 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  34 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  35
  36 import java.util.concurrent.ExecutionException;
  37
  38 /**
  39  * The ShardTransaction Actor represents a remote transaction
  40  *
  41  * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
  42  *
  43  * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
  44  * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
  45  * time there are no known advantages for creating a read-only or write-only transaction which may change over time
  46  * at which point we can optimize things in the distributed store as well.
  47  *
  48  * Handles Messages
  49  * ----------------
  50  * {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
  51  * {@link org.opendaylight.controller.cluster.datastore.messages.WriteData}
  52  * {@link org.opendaylight.controller.cluster.datastore.messages.MergeData}
  53  * {@link org.opendaylight.controller.cluster.datastore.messages.DeleteData}
  54  * {@link org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction}
  55  * {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
  56  */
  57 public class ShardTransaction extends UntypedActor {
  58
  59   private final DOMStoreReadWriteTransaction transaction;
  60
  61   private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  62
  63   public ShardTransaction(DOMStoreReadWriteTransaction transaction) {
  64     this.transaction = transaction;
  65   }
  66
  67
  68   public static Props props(final DOMStoreReadWriteTransaction transaction){
  69     return Props.create(new Creator<ShardTransaction>(){
  70
  71       @Override
  72       public ShardTransaction create() throws Exception {
  73         return new ShardTransaction(transaction);
  74       }
  75     });
  76   }
  77
  78   @Override
  79   public void onReceive(Object message) throws Exception {
  80     if(message instanceof ReadData){
  81       readData((ReadData) message);
  82     } else if(message instanceof WriteData){
  83       writeData((WriteData) message);
  84     } else if(message instanceof MergeData){
  85       mergeData((MergeData) message);
  86     } else if(message instanceof DeleteData){
  87       deleteData((DeleteData) message);
  88     } else if(message instanceof ReadyTransaction){
  89       readyTransaction((ReadyTransaction) message);
  90     } else if(message instanceof CloseTransaction){
  91       closeTransaction((CloseTransaction) message);
  92     }
  93   }
  94
  95   private void readData(ReadData message) {
  96     final ActorRef sender = getSender();
  97     final ActorRef self = getSelf();
  98     final InstanceIdentifier path = message.getPath();
  99     final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = transaction.read(path);
 100
 101     future.addListener(new Runnable() {
 102       @Override
 103       public void run() {
 104         try {
 105           Optional<NormalizedNode<?, ?>> optional = future.get();
 106           if(optional.isPresent()){
 107             sender.tell(new ReadDataReply(optional.get()), self);
 108           } else {
 109             //TODO : Need to decide what to do here
 110           }
 111         } catch (InterruptedException | ExecutionException e) {
 112           log.error(e, "An exception happened when reading data from path : " + path.toString());
 113         }
 114
 115       }
 116     }, getContext().dispatcher());
 117   }
 118
 119
 120   private void writeData(WriteData message){
 121     transaction.write(message.getPath(), message.getData());
 122     getSender().tell(new WriteDataReply(), getSelf());
 123   }
 124
 125   private void mergeData(MergeData message){
 126     transaction.merge(message.getPath(), message.getData());
 127     getSender().tell(new MergeDataReply(), getSelf());
 128   }
 129
 130   private void deleteData(DeleteData message){
 131     transaction.delete(message.getPath());
 132     getSender().tell(new DeleteDataReply(), getSelf());
 133   }
 134
 135   private void readyTransaction(ReadyTransaction message){
 136     DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
 137     ActorRef cohortActor = getContext().actorOf(ThreePhaseCommitCohort.props(cohort));
 138     getSender().tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
 139
 140   }
 141
 142   private void closeTransaction(CloseTransaction message){
 143     transaction.close();
 144     getSender().tell(new CloseTransactionReply(), getSelf());
 145   }
 146 }