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%2FShardTransactionChain.java;h=aedc6c42b862d0cc2ffd94fc9082b82d38f2aa2e;hb=30507b196fa240a4176ba12102ac0469280feff9;hp=83913fe416fb766698ce9bd4294819d48276905d;hpb=bef0749bb2517eaae6a501be501e16a7d3905045;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransactionChain.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransactionChain.java index 83913fe416..aedc6c42b8 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransactionChain.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransactionChain.java @@ -8,47 +8,103 @@ package org.opendaylight.controller.cluster.datastore; +import com.google.common.base.Preconditions; import akka.actor.ActorRef; import akka.actor.Props; -import akka.actor.UntypedActor; import akka.japi.Creator; +import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor; +import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChainReply; import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; +import org.opendaylight.yangtools.yang.model.api.SchemaContext; /** * The ShardTransactionChain Actor represents a remote TransactionChain */ -public class ShardTransactionChain extends UntypedActor{ - - private final DOMStoreTransactionChain chain; - - public ShardTransactionChain(DOMStoreTransactionChain chain) { - this.chain = chain; - } - - @Override - public void onReceive(Object message) throws Exception { - if(message instanceof CreateTransaction){ - DOMStoreReadWriteTransaction transaction = chain.newReadWriteTransaction(); - ActorRef transactionActor = getContext().actorOf(ShardTransaction.props(transaction)); - getSender().tell(new CreateTransactionReply(transactionActor.path()), getSelf()); - } else if (message instanceof CloseTransactionChain){ - chain.close(); - getSender().tell(new CloseTransactionChainReply(), getSelf()); +public class ShardTransactionChain extends AbstractUntypedActor { + + private final ShardDataTreeTransactionChain chain; + private final DatastoreContext datastoreContext; + private final ShardStats shardStats; + + public ShardTransactionChain(ShardDataTreeTransactionChain chain, DatastoreContext datastoreContext, + ShardStats shardStats) { + this.chain = Preconditions.checkNotNull(chain); + this.datastoreContext = datastoreContext; + this.shardStats = shardStats; + } + + @Override + public void handleReceive(Object message) throws Exception { + if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) { + CreateTransaction createTransaction = CreateTransaction.fromSerializable( message); + createTransaction(createTransaction); + } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)) { + chain.close(); + getSender().tell(CloseTransactionChainReply.INSTANCE.toSerializable(), getSelf()); + }else{ + unknownMessage(message); + } + } + + private ActorRef getShardActor(){ + return getContext().parent(); + } + + private ActorRef createTypedTransactionActor(CreateTransaction createTransaction) { + String transactionName = "shard-" + createTransaction.getTransactionId(); + + final TransactionType type = TransactionType.fromInt(createTransaction.getTransactionType()); + final AbstractShardDataTreeTransaction transaction; + switch (type) { + case READ_ONLY: + transaction = chain.newReadOnlyTransaction(transactionName); + break; + case READ_WRITE: + case WRITE_ONLY: + transaction = chain.newReadWriteTransaction(transactionName); + break; + default: + throw new IllegalArgumentException("Unhandled transaction type " + type); + } + + return getContext().actorOf( + ShardTransaction.props(type, transaction, getShardActor(), + datastoreContext, shardStats, createTransaction.getTransactionId(), + createTransaction.getVersion()), transactionName); } - } - public static Props props(final DOMStoreTransactionChain chain){ - return Props.create(new Creator(){ + private void createTransaction(CreateTransaction createTransaction) { + + ActorRef transactionActor = createTypedTransactionActor(createTransaction); + getSender().tell(new CreateTransactionReply(transactionActor.path().toString(), + createTransaction.getTransactionId()).toSerializable(), getSelf()); + } + + public static Props props(ShardDataTreeTransactionChain chain, SchemaContext schemaContext, + DatastoreContext datastoreContext, ShardStats shardStats) { + return Props.create(new ShardTransactionChainCreator(chain, datastoreContext, shardStats)); + } - @Override - public ShardTransactionChain create() throws Exception { - return new ShardTransactionChain(chain); - } - }); - } + private static class ShardTransactionChainCreator implements Creator { + private static final long serialVersionUID = 1L; + + final ShardDataTreeTransactionChain chain; + final DatastoreContext datastoreContext; + final ShardStats shardStats; + + ShardTransactionChainCreator(ShardDataTreeTransactionChain chain, DatastoreContext datastoreContext, + ShardStats shardStats) { + this.chain = chain; + this.datastoreContext = datastoreContext; + this.shardStats = shardStats; + } + + @Override + public ShardTransactionChain create() throws Exception { + return new ShardTransactionChain(chain, datastoreContext, shardStats); + } + } }