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.PoisonPill;
  13 import akka.actor.Props;
  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.controller.sal.core.spi.data.DOMStoreTransactionChain;
  40 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  42
  43 import java.util.concurrent.ExecutionException;
  44
  45 /**
  46  * The ShardTransaction Actor represents a remote transaction
  47  * <p>
  48  * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
  49  * </p>
  50  * <p>
  51  * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
  52  * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
  53  * time there are no known advantages for creating a read-only or write-only transaction which may change over time
  54  * at which point we can optimize things in the distributed store as well.
  55  * </p>
  56  * <p>
  57  * Handles Messages <br/>
  58  * ---------------- <br/>
  59  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
  60  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.WriteData}
  61  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.MergeData}
  62  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.DeleteData}
  63  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction}
  64  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
  65  * </p>
  66  */
  67 public class ShardTransaction extends AbstractUntypedActor {
  68
  69     private final ActorRef shardActor;
  70
  71     // FIXME : see below
  72     // If transactionChain is not null then this transaction is part of a
  73     // transactionChain. Not really clear as to what that buys us
  74     private final DOMStoreTransactionChain transactionChain;
  75
  76     private final DOMStoreReadWriteTransaction transaction;
  77
  78     private final MutableCompositeModification modification =
  79         new MutableCompositeModification();
  80
  81     private final LoggingAdapter log =
  82         Logging.getLogger(getContext().system(), this);
  83
  84     public ShardTransaction(DOMStoreReadWriteTransaction transaction,
  85         ActorRef shardActor) {
  86         this(null, transaction, shardActor);
  87     }
  88
  89     public ShardTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadWriteTransaction transaction,
  90         ActorRef shardActor) {
  91         this.transactionChain = transactionChain;
  92         this.transaction = transaction;
  93         this.shardActor = shardActor;
  94     }
  95
  96
  97
  98     public static Props props(final DOMStoreReadWriteTransaction transaction,
  99         final ActorRef shardActor) {
 100         return Props.create(new Creator<ShardTransaction>() {
 101
 102             @Override
 103             public ShardTransaction create() throws Exception {
 104                 return new ShardTransaction(transaction, shardActor);
 105             }
 106         });
 107     }
 108
 109     public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
 110         final ActorRef shardActor) {
 111         return Props.create(new Creator<ShardTransaction>() {
 112
 113             @Override
 114             public ShardTransaction create() throws Exception {
 115                 return new ShardTransaction(transactionChain, transaction, shardActor);
 116             }
 117         });
 118     }
 119
 120
 121     @Override
 122     public void handleReceive(Object message) throws Exception {
 123         if (message instanceof ReadData) {
 124             readData((ReadData) message);
 125         } else if (message instanceof WriteData) {
 126             writeData((WriteData) message);
 127         } else if (message instanceof MergeData) {
 128             mergeData((MergeData) message);
 129         } else if (DeleteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
 130             deleteData(DeleteData.fromSerizalizable(message));
 131         } else if (message instanceof ReadyTransaction) {
 132             readyTransaction((ReadyTransaction) message);
 133         } else if (message instanceof CloseTransaction) {
 134             closeTransaction((CloseTransaction) message);
 135         } else if (message instanceof GetCompositedModification) {
 136             // This is here for testing only
 137             getSender().tell(new GetCompositeModificationReply(
 138                 new ImmutableCompositeModification(modification)), getSelf());
 139         }
 140     }
 141
 142     private void readData(ReadData message) {
 143         final ActorRef sender = getSender();
 144         final ActorRef self = getSelf();
 145         final InstanceIdentifier path = message.getPath();
 146         final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
 147             transaction.read(path);
 148
 149         future.addListener(new Runnable() {
 150             @Override
 151             public void run() {
 152                 try {
 153                     Optional<NormalizedNode<?, ?>> optional = future.get();
 154                     if (optional.isPresent()) {
 155                         sender.tell(new ReadDataReply(optional.get()), self);
 156                     } else {
 157                         sender.tell(new ReadDataReply(null), self);
 158                     }
 159                 } catch (InterruptedException | ExecutionException e) {
 160                     log.error(e,
 161                         "An exception happened when reading data from path : "
 162                             + path.toString());
 163                 }
 164
 165             }
 166         }, getContext().dispatcher());
 167     }
 168
 169
 170     private void writeData(WriteData message) {
 171         modification.addModification(
 172             new WriteModification(message.getPath(), message.getData()));
 173         transaction.write(message.getPath(), message.getData());
 174         getSender().tell(new WriteDataReply(), getSelf());
 175     }
 176
 177     private void mergeData(MergeData message) {
 178         modification.addModification(
 179             new MergeModification(message.getPath(), message.getData()));
 180         transaction.merge(message.getPath(), message.getData());
 181         getSender().tell(new MergeDataReply(), getSelf());
 182     }
 183
 184     private void deleteData(DeleteData message) {
 185         modification.addModification(new DeleteModification(message.getPath()));
 186         transaction.delete(message.getPath());
 187         getSender().tell(new DeleteDataReply(), getSelf());
 188     }
 189
 190     private void readyTransaction(ReadyTransaction message) {
 191         DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
 192         ActorRef cohortActor = getContext().actorOf(
 193             ThreePhaseCommitCohort.props(cohort, shardActor, modification), "cohort");
 194         getSender()
 195             .tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
 196
 197     }
 198
 199     private void closeTransaction(CloseTransaction message) {
 200         transaction.close();
 201         getSender().tell(new CloseTransactionReply(), getSelf());
 202         getSelf().tell(PoisonPill.getInstance(), getSelf());
 203     }
 204
 205
 206     // These classes are in here for test purposes only
 207
 208
 209     static class GetCompositedModification {
 210
 211     }
 212
 213
 214     static class GetCompositeModificationReply {
 215         private final CompositeModification modification;
 216
 217
 218         GetCompositeModificationReply(CompositeModification modification) {
 219             this.modification = modification;
 220         }
 221
 222
 223         public CompositeModification getModification() {
 224             return modification;
 225         }
 226     }
 227 }