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.YangInstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  42 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  43
  44 import java.util.concurrent.ExecutionException;
  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 class ShardTransaction extends AbstractUntypedActor {
  69
  70     private final ActorRef shardActor;
  71     private 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     private final DOMStoreReadWriteTransaction transaction;
  79
  80     private final MutableCompositeModification modification =
  81         new MutableCompositeModification();
  82
  83     private final LoggingAdapter log =
  84         Logging.getLogger(getContext().system(), this);
  85
  86     public ShardTransaction(DOMStoreReadWriteTransaction transaction,
  87         ActorRef shardActor, SchemaContext schemaContext) {
  88         this(null, transaction, shardActor, schemaContext);
  89     }
  90
  91     public ShardTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadWriteTransaction transaction,
  92         ActorRef shardActor, SchemaContext schemaContext) {
  93         this.transactionChain = transactionChain;
  94         this.transaction = transaction;
  95         this.shardActor = shardActor;
  96         this.schemaContext = schemaContext;
  97     }
  98
  99
 100
 101     public static Props props(final DOMStoreReadWriteTransaction transaction,
 102         final ActorRef shardActor, final SchemaContext schemaContext) {
 103         return Props.create(new Creator<ShardTransaction>() {
 104
 105             @Override
 106             public ShardTransaction create() throws Exception {
 107                 return new ShardTransaction(transaction, shardActor, schemaContext);
 108             }
 109         });
 110     }
 111
 112     public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
 113         final ActorRef shardActor, final SchemaContext schemaContext) {
 114         return Props.create(new Creator<ShardTransaction>() {
 115
 116             @Override
 117             public ShardTransaction create() throws Exception {
 118                 return new ShardTransaction(transactionChain, transaction, shardActor, schemaContext);
 119             }
 120         });
 121     }
 122
 123
 124     @Override
 125     public void handleReceive(Object message) throws Exception {
 126         if (ReadData.SERIALIZABLE_CLASS.equals(message.getClass())) {
 127             readData(ReadData.fromSerializable(message));
 128         } else if (WriteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
 129             writeData(WriteData.fromSerializable(message, schemaContext));
 130         } else if (MergeData.SERIALIZABLE_CLASS.equals(message.getClass())) {
 131             mergeData(MergeData.fromSerializable(message, schemaContext));
 132         } else if (DeleteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
 133             deleteData(DeleteData.fromSerizalizable(message));
 134         } else if (ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
 135             readyTransaction(new ReadyTransaction());
 136         } else if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) {
 137             closeTransaction(new CloseTransaction());
 138         } else if (message instanceof GetCompositedModification) {
 139             // This is here for testing only
 140             getSender().tell(new GetCompositeModificationReply(
 141                 new ImmutableCompositeModification(modification)), getSelf());
 142         }else{
 143           throw new Exception ("Shard:handleRecieve received an unknown message"+message);
 144         }
 145     }
 146
 147     private void readData(ReadData message) {
 148         final ActorRef sender = getSender();
 149         final ActorRef self = getSelf();
 150         final YangInstanceIdentifier path = message.getPath();
 151         final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
 152             transaction.read(path);
 153
 154         future.addListener(new Runnable() {
 155             @Override
 156             public void run() {
 157                 try {
 158                     Optional<NormalizedNode<?, ?>> optional = future.get();
 159                     if (optional.isPresent()) {
 160                         sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
 161                     } else {
 162                         sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self);
 163                     }
 164                 } catch (InterruptedException | ExecutionException e) {
 165                     log.error(e,
 166                         "An exception happened when reading data from path : "
 167                             + path.toString());
 168                 }
 169
 170             }
 171         }, getContext().dispatcher());
 172     }
 173
 174
 175     private void writeData(WriteData message) {
 176         modification.addModification(
 177             new WriteModification(message.getPath(), message.getData(),schemaContext));
 178         LOG.debug("writeData at path : " + message.getPath().toString());
 179         transaction.write(message.getPath(), message.getData());
 180         getSender().tell(new WriteDataReply().toSerializable(), getSelf());
 181     }
 182
 183     private void mergeData(MergeData message) {
 184         modification.addModification(
 185             new MergeModification(message.getPath(), message.getData(), schemaContext));
 186         LOG.debug("mergeData at path : " + message.getPath().toString());
 187         transaction.merge(message.getPath(), message.getData());
 188         getSender().tell(new MergeDataReply().toSerializable(), getSelf());
 189     }
 190
 191     private void deleteData(DeleteData message) {
 192         modification.addModification(new DeleteModification(message.getPath()));
 193         transaction.delete(message.getPath());
 194         getSender().tell(new DeleteDataReply().toSerializable(), getSelf());
 195     }
 196
 197     private void readyTransaction(ReadyTransaction message) {
 198         DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
 199         ActorRef cohortActor = getContext().actorOf(
 200             ThreePhaseCommitCohort.props(cohort, shardActor, modification), "cohort");
 201         getSender()
 202             .tell(new ReadyTransactionReply(cohortActor.path()).toSerializable(), getSelf());
 203
 204     }
 205
 206     private void closeTransaction(CloseTransaction message) {
 207         transaction.close();
 208         getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
 209         getSelf().tell(PoisonPill.getInstance(), getSelf());
 210     }
 211
 212
 213     // These classes are in here for test purposes only
 214
 215
 216     static class GetCompositedModification {
 217
 218     }
 219
 220
 221     static class GetCompositeModificationReply {
 222         private final CompositeModification modification;
 223
 224
 225         GetCompositeModificationReply(CompositeModification modification) {
 226             this.modification = modification;
 227         }
 228
 229
 230         public CompositeModification getModification() {
 231             return modification;
 232         }
 233     }
 234 }