X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardTransaction.java;h=1ffe5ca40262ef630a247dca0ce8e46e126f3c34;hb=886fe1a50077d9dc9c4c36f938fc7c86317cb149;hp=75744cad5b920b942efda865d0fa6ed29c41f3d7;hpb=f78020d87663c8d9db1e4e33939f7b8b703703f8;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java index 75744cad5b..1ffe5ca402 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java @@ -10,14 +10,13 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorRef; import akka.actor.Props; -import akka.actor.UntypedActor; import akka.event.Logging; import akka.event.LoggingAdapter; import akka.japi.Creator; import com.google.common.base.Optional; import com.google.common.util.concurrent.ListenableFuture; +import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; -import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.DeleteData; import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply; import org.opendaylight.controller.cluster.datastore.messages.MergeData; @@ -34,25 +33,28 @@ import org.opendaylight.controller.cluster.datastore.modification.ImmutableCompo import org.opendaylight.controller.cluster.datastore.modification.MergeModification; import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; import org.opendaylight.controller.cluster.datastore.modification.WriteModification; +import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; +import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier; +import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; +import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; - -import java.util.concurrent.ExecutionException; +import org.opendaylight.yangtools.yang.model.api.SchemaContext; /** * The ShardTransaction Actor represents a remote transaction - *

+ *

* The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction - *

- *

+ *

+ *

* Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this * time there are no known advantages for creating a read-only or write-only transaction which may change over time * at which point we can optimize things in the distributed store as well. - *

- *

+ *

+ *

* Handles Messages
* ----------------
*

  • {@link org.opendaylight.controller.cluster.datastore.messages.ReadData} @@ -63,70 +65,140 @@ import java.util.concurrent.ExecutionException; *
  • {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction} *

    */ -public class ShardTransaction extends UntypedActor { +public abstract class ShardTransaction extends AbstractUntypedActor { private final ActorRef shardActor; + protected final SchemaContext schemaContext; + + // FIXME : see below + // If transactionChain is not null then this transaction is part of a + // transactionChain. Not really clear as to what that buys us + private final DOMStoreTransactionChain transactionChain; - private final DOMStoreReadWriteTransaction transaction; - private final MutableCompositeModification modification = new MutableCompositeModification(); + private final MutableCompositeModification modification = + new MutableCompositeModification(); - private final LoggingAdapter log = Logging.getLogger(getContext().system(), this); + private final LoggingAdapter log = + Logging.getLogger(getContext().system(), this); - public ShardTransaction(DOMStoreReadWriteTransaction transaction, ActorRef shardActor) { - this.transaction = transaction; + protected ShardTransaction( + ActorRef shardActor, SchemaContext schemaContext) { + this(null, shardActor, schemaContext); + } + + protected ShardTransaction(DOMStoreTransactionChain transactionChain, + ActorRef shardActor, SchemaContext schemaContext) { + this.transactionChain = transactionChain; this.shardActor = shardActor; + this.schemaContext = schemaContext; } - public static Props props(final DOMStoreReadWriteTransaction transaction, final ActorRef shardActor){ - return Props.create(new Creator(){ + + public static Props props(final DOMStoreReadTransaction transaction, + final ActorRef shardActor, final SchemaContext schemaContext) { + return Props.create(new Creator() { + + @Override + public ShardTransaction create() throws Exception { + return new ShardReadTransaction(transaction, shardActor, schemaContext); + } + }); + } + + public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadTransaction transaction, + final ActorRef shardActor, final SchemaContext schemaContext) { + return Props.create(new Creator() { @Override public ShardTransaction create() throws Exception { - return new ShardTransaction(transaction, shardActor); + return new ShardReadTransaction(transactionChain, transaction, shardActor, schemaContext); } }); } + public static Props props(final DOMStoreReadWriteTransaction transaction, + final ActorRef shardActor, final SchemaContext schemaContext) { + return Props.create(new Creator() { + + @Override + public ShardTransaction create() throws Exception { + return new ShardReadWriteTransaction(transaction, shardActor, schemaContext); + } + }); + } + + public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction, + final ActorRef shardActor, final SchemaContext schemaContext) { + return Props.create(new Creator() { + + @Override + public ShardTransaction create() throws Exception { + return new ShardReadWriteTransaction(transactionChain, transaction, shardActor, schemaContext); + } + }); + } + + + public static Props props(final DOMStoreWriteTransaction transaction, + final ActorRef shardActor, final SchemaContext schemaContext) { + return Props.create(new Creator() { + + @Override + public ShardTransaction create() throws Exception { + return new ShardWriteTransaction(transaction, shardActor, schemaContext); + } + }); + } + + public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreWriteTransaction transaction, + final ActorRef shardActor, final SchemaContext schemaContext) { + return Props.create(new Creator() { + + @Override + public ShardTransaction create() throws Exception { + return new ShardWriteTransaction(transactionChain, transaction, shardActor, schemaContext); + } + }); + } + + @Override - public void onReceive(Object message) throws Exception { - if(message instanceof ReadData){ - readData((ReadData) message); - } else if(message instanceof WriteData){ - writeData((WriteData) message); - } else if(message instanceof MergeData){ - mergeData((MergeData) message); - } else if(message instanceof DeleteData){ - deleteData((DeleteData) message); - } else if(message instanceof ReadyTransaction){ - readyTransaction((ReadyTransaction) message); - } else if(message instanceof CloseTransaction){ - closeTransaction((CloseTransaction) message); - } else if(message instanceof GetCompositedModification){ + public void handleReceive(Object message) throws Exception { + if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) { + closeTransaction(new CloseTransaction()); + } else if (message instanceof GetCompositedModification) { // This is here for testing only - getSender().tell(new GetCompositeModificationReply(new ImmutableCompositeModification(modification)), getSelf()); + getSender().tell(new GetCompositeModificationReply( + new ImmutableCompositeModification(modification)), getSelf()); + }else{ + throw new UnknownMessageException(message); } } - private void readData(ReadData message) { + abstract protected void closeTransaction(CloseTransaction message); + + protected void readData(DOMStoreReadTransaction transaction,ReadData message) { final ActorRef sender = getSender(); final ActorRef self = getSelf(); - final InstanceIdentifier path = message.getPath(); - final ListenableFuture>> future = transaction.read(path); + final YangInstanceIdentifier path = message.getPath(); + final ListenableFuture>> future = + transaction.read(path); future.addListener(new Runnable() { @Override public void run() { try { Optional> optional = future.get(); - if(optional.isPresent()){ - sender.tell(new ReadDataReply(optional.get()), self); + if (optional.isPresent()) { + sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self); } else { - //TODO : Need to decide what to do here + sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self); } - } catch (InterruptedException | ExecutionException e) { - log.error(e, "An exception happened when reading data from path : " + path.toString()); + } catch (Exception e) { + sender.tell(new akka.actor.Status.Failure(new ReadFailedException( "An Exception occurred when reading data from path : " + + path.toString(),e)),self); } } @@ -134,43 +206,60 @@ public class ShardTransaction extends UntypedActor { } - private void writeData(WriteData message){ - modification.addModification(new WriteModification(message.getPath(), message.getData())); - transaction.write(message.getPath(), message.getData()); - getSender().tell(new WriteDataReply(), getSelf()); + protected void writeData(DOMStoreWriteTransaction transaction, WriteData message) { + modification.addModification( + new WriteModification(message.getPath(), message.getData(),schemaContext)); + LOG.debug("writeData at path : " + message.getPath().toString()); + + try { + transaction.write(message.getPath(), message.getData()); + getSender().tell(new WriteDataReply().toSerializable(), getSelf()); + }catch(Exception e){ + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } } - private void mergeData(MergeData message){ - modification.addModification(new MergeModification(message.getPath(), message.getData())); - transaction.merge(message.getPath(), message.getData()); - getSender().tell(new MergeDataReply(), getSelf()); + protected void mergeData(DOMStoreWriteTransaction transaction, MergeData message) { + modification.addModification( + new MergeModification(message.getPath(), message.getData(), schemaContext)); + LOG.debug("mergeData at path : " + message.getPath().toString()); + try { + transaction.merge(message.getPath(), message.getData()); + getSender().tell(new MergeDataReply().toSerializable(), getSelf()); + }catch(Exception e){ + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } } - private void deleteData(DeleteData message){ + protected void deleteData(DOMStoreWriteTransaction transaction, DeleteData message) { + LOG.debug("deleteData at path : " + message.getPath().toString()); modification.addModification(new DeleteModification(message.getPath())); - transaction.delete(message.getPath()); - getSender().tell(new DeleteDataReply(), getSelf()); + try { + transaction.delete(message.getPath()); + getSender().tell(new DeleteDataReply().toSerializable(), getSelf()); + }catch(Exception e){ + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } } - private void readyTransaction(ReadyTransaction message){ - DOMStoreThreePhaseCommitCohort cohort = transaction.ready(); - ActorRef cohortActor = getContext().actorOf(ThreePhaseCommitCohort.props(cohort, shardActor, modification)); - getSender().tell(new ReadyTransactionReply(cohortActor.path()), getSelf()); - - } + protected void readyTransaction(DOMStoreWriteTransaction transaction, ReadyTransaction message) { + DOMStoreThreePhaseCommitCohort cohort = transaction.ready(); + ActorRef cohortActor = getContext().actorOf( + ThreePhaseCommitCohort.props(cohort, shardActor, modification), "cohort"); + getSender() + .tell(new ReadyTransactionReply(cohortActor.path()).toSerializable(), getSelf()); - private void closeTransaction(CloseTransaction message){ - transaction.close(); - getSender().tell(new CloseTransactionReply(), getSelf()); } // These classes are in here for test purposes only + static class GetCompositedModification { } + static class GetCompositeModificationReply { private final CompositeModification modification;