*/
com.google.protobuf.ByteString
getTransactionIdBytes();
+
+ // required int32 transactionType = 2;
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ boolean hasTransactionType();
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ int getTransactionType();
}
/**
* Protobuf type {@code org.opendaylight.controller.mdsal.CreateTransaction}
transactionId_ = input.readBytes();
break;
}
+ case 16: {
+ bitField0_ |= 0x00000002;
+ transactionType_ = input.readInt32();
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
}
}
+ // required int32 transactionType = 2;
+ public static final int TRANSACTIONTYPE_FIELD_NUMBER = 2;
+ private int transactionType_;
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ public boolean hasTransactionType() {
+ return ((bitField0_ & 0x00000002) == 0x00000002);
+ }
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ public int getTransactionType() {
+ return transactionType_;
+ }
+
private void initFields() {
transactionId_ = "";
+ transactionType_ = 0;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
memoizedIsInitialized = 0;
return false;
}
+ if (!hasTransactionType()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
memoizedIsInitialized = 1;
return true;
}
if (((bitField0_ & 0x00000001) == 0x00000001)) {
output.writeBytes(1, getTransactionIdBytes());
}
+ if (((bitField0_ & 0x00000002) == 0x00000002)) {
+ output.writeInt32(2, transactionType_);
+ }
getUnknownFields().writeTo(output);
}
size += com.google.protobuf.CodedOutputStream
.computeBytesSize(1, getTransactionIdBytes());
}
+ if (((bitField0_ & 0x00000002) == 0x00000002)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeInt32Size(2, transactionType_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
super.clear();
transactionId_ = "";
bitField0_ = (bitField0_ & ~0x00000001);
+ transactionType_ = 0;
+ bitField0_ = (bitField0_ & ~0x00000002);
return this;
}
to_bitField0_ |= 0x00000001;
}
result.transactionId_ = transactionId_;
+ if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
+ to_bitField0_ |= 0x00000002;
+ }
+ result.transactionType_ = transactionType_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
transactionId_ = other.transactionId_;
onChanged();
}
+ if (other.hasTransactionType()) {
+ setTransactionType(other.getTransactionType());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
return false;
}
+ if (!hasTransactionType()) {
+
+ return false;
+ }
return true;
}
return this;
}
+ // required int32 transactionType = 2;
+ private int transactionType_ ;
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ public boolean hasTransactionType() {
+ return ((bitField0_ & 0x00000002) == 0x00000002);
+ }
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ public int getTransactionType() {
+ return transactionType_;
+ }
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ public Builder setTransactionType(int value) {
+ bitField0_ |= 0x00000002;
+ transactionType_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>required int32 transactionType = 2;</code>
+ */
+ public Builder clearTransactionType() {
+ bitField0_ = (bitField0_ & ~0x00000002);
+ transactionType_ = 0;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:org.opendaylight.controller.mdsal.CreateTransaction)
}
java.lang.String[] descriptorData = {
"\n\026ShardTransaction.proto\022!org.opendaylig" +
"ht.controller.mdsal\032\014Common.proto\"\022\n\020Clo" +
- "seTransaction\"\027\n\025CloseTransactionReply\"*" +
+ "seTransaction\"\027\n\025CloseTransactionReply\"C" +
"\n\021CreateTransaction\022\025\n\rtransactionId\030\001 \002" +
- "(\t\"M\n\026CreateTransactionReply\022\034\n\024transact" +
- "ionActorPath\030\001 \002(\t\022\025\n\rtransactionId\030\002 \002(" +
- "\t\"\022\n\020ReadyTransaction\"*\n\025ReadyTransactio" +
- "nReply\022\021\n\tactorPath\030\001 \002(\t\"l\n\nDeleteData\022" +
- "^\n\037instanceIdentifierPathArguments\030\001 \002(\013" +
- "25.org.opendaylight.controller.mdsal.Ins",
- "tanceIdentifier\"\021\n\017DeleteDataReply\"j\n\010Re" +
- "adData\022^\n\037instanceIdentifierPathArgument" +
- "s\030\001 \002(\01325.org.opendaylight.controller.md" +
- "sal.InstanceIdentifier\"P\n\rReadDataReply\022" +
- "?\n\016normalizedNode\030\001 \001(\0132\'.org.opendaylig" +
- "ht.controller.mdsal.Node\"\254\001\n\tWriteData\022^" +
- "\n\037instanceIdentifierPathArguments\030\001 \002(\0132" +
- "5.org.opendaylight.controller.mdsal.Inst" +
- "anceIdentifier\022?\n\016normalizedNode\030\002 \002(\0132\'" +
- ".org.opendaylight.controller.mdsal.Node\"",
- "\020\n\016WriteDataReply\"\254\001\n\tMergeData\022^\n\037insta" +
+ "(\t\022\027\n\017transactionType\030\002 \002(\005\"M\n\026CreateTra" +
+ "nsactionReply\022\034\n\024transactionActorPath\030\001 " +
+ "\002(\t\022\025\n\rtransactionId\030\002 \002(\t\"\022\n\020ReadyTrans" +
+ "action\"*\n\025ReadyTransactionReply\022\021\n\tactor" +
+ "Path\030\001 \002(\t\"l\n\nDeleteData\022^\n\037instanceIden" +
+ "tifierPathArguments\030\001 \002(\01325.org.opendayl",
+ "ight.controller.mdsal.InstanceIdentifier" +
+ "\"\021\n\017DeleteDataReply\"j\n\010ReadData\022^\n\037insta" +
"nceIdentifierPathArguments\030\001 \002(\01325.org.o" +
"pendaylight.controller.mdsal.InstanceIde" +
- "ntifier\022?\n\016normalizedNode\030\002 \002(\0132\'.org.op" +
- "endaylight.controller.mdsal.Node\"\020\n\016Merg" +
- "eDataReplyBV\n:org.opendaylight.controlle" +
- "r.protobuff.messages.transactionB\030ShardT" +
- "ransactionMessages"
+ "ntifier\"P\n\rReadDataReply\022?\n\016normalizedNo" +
+ "de\030\001 \001(\0132\'.org.opendaylight.controller.m" +
+ "dsal.Node\"\254\001\n\tWriteData\022^\n\037instanceIdent" +
+ "ifierPathArguments\030\001 \002(\01325.org.opendayli" +
+ "ght.controller.mdsal.InstanceIdentifier\022" +
+ "?\n\016normalizedNode\030\002 \002(\0132\'.org.opendaylig",
+ "ht.controller.mdsal.Node\"\020\n\016WriteDataRep" +
+ "ly\"\254\001\n\tMergeData\022^\n\037instanceIdentifierPa" +
+ "thArguments\030\001 \002(\01325.org.opendaylight.con" +
+ "troller.mdsal.InstanceIdentifier\022?\n\016norm" +
+ "alizedNode\030\002 \002(\0132\'.org.opendaylight.cont" +
+ "roller.mdsal.Node\"\020\n\016MergeDataReplyBV\n:o" +
+ "rg.opendaylight.controller.protobuff.mes" +
+ "sages.transactionB\030ShardTransactionMessa" +
+ "ges"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
internal_static_org_opendaylight_controller_mdsal_CreateTransaction_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_org_opendaylight_controller_mdsal_CreateTransaction_descriptor,
- new java.lang.String[] { "TransactionId", });
+ new java.lang.String[] { "TransactionId", "TransactionType", });
internal_static_org_opendaylight_controller_mdsal_CreateTransactionReply_descriptor =
getDescriptor().getMessageTypes().get(3);
internal_static_org_opendaylight_controller_mdsal_CreateTransactionReply_fieldAccessorTable = new
message CreateTransaction{
required string transactionId = 1;
+ required int32 transactionType =2;
}
message CreateTransactionReply{
import akka.event.LoggingAdapter;
import akka.japi.Creator;
import akka.serialization.Serialization;
-
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
-
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
}
}
+ private ActorRef createTypedTransactionActor(CreateTransaction createTransaction,String transactionId){
+ if(createTransaction.getTransactionType()== TransactionProxy.TransactionType.READ_ONLY.ordinal()){
+ return getContext().actorOf(
+ ShardTransaction.props( store.newReadOnlyTransaction(), getSelf(), schemaContext), transactionId);
+
+ }else if (createTransaction.getTransactionType()== TransactionProxy.TransactionType.READ_WRITE.ordinal()){
+ return getContext().actorOf(
+ ShardTransaction.props( store.newReadWriteTransaction(), getSelf(), schemaContext), transactionId);
+
+
+ }else if (createTransaction.getTransactionType()== TransactionProxy.TransactionType.WRITE_ONLY.ordinal()){
+ return getContext().actorOf(
+ ShardTransaction.props( store.newWriteOnlyTransaction(), getSelf(), schemaContext), transactionId);
+ }else{
+ throw new IllegalArgumentException ("CreateTransaction message has unidentified transaction type="+createTransaction.getTransactionType()) ;
+ }
+ }
+
private void createTransaction(CreateTransaction createTransaction) {
- DOMStoreReadWriteTransaction transaction =
- store.newReadWriteTransaction();
+
String transactionId = "shard-" + createTransaction.getTransactionId();
LOG.info("Creating transaction : {} " , transactionId);
- ActorRef transactionActor = getContext().actorOf(
- ShardTransaction.props(transaction, getSelf(), schemaContext), transactionId);
+ ActorRef transactionActor = createTypedTransactionActor(createTransaction,transactionId);
getSender()
.tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), createTransaction.getTransactionId()).toSerializable(),
--- /dev/null
+/*
+ *
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.ReadData;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+
+/**
+ * @author: syedbahm
+ * Date: 8/6/14
+ */
+public class ShardReadTransaction extends ShardTransaction {
+ private final DOMStoreReadTransaction transaction;
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+
+ public ShardReadTransaction(DOMStoreReadTransaction transaction, ActorRef shardActor, SchemaContext schemaContext) {
+ super(shardActor, schemaContext);
+ this.transaction = transaction;
+
+ }
+
+ public ShardReadTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadTransaction transaction, ActorRef shardActor, SchemaContext schemaContext) {
+ super(transactionChain, shardActor, schemaContext);
+ this.transaction = transaction;
+ }
+
+ @Override
+ public void handleReceive(Object message) throws Exception {
+ if (ReadData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ readData(transaction,ReadData.fromSerializable(message));
+ } else {
+ super.handleReceive(message);
+ }
+ }
+ protected void closeTransaction(CloseTransaction message) {
+ transaction.close();
+ getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
+ getSelf().tell(PoisonPill.getInstance(), getSelf());
+ }
+
+}
--- /dev/null
+/*
+ *
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+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.MergeData;
+import org.opendaylight.controller.cluster.datastore.messages.ReadData;
+import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.WriteData;
+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;
+
+/**
+ * @author: syedbahm
+ * Date: 8/6/14
+ */
+public class ShardReadWriteTransaction extends ShardTransaction {
+ private final DOMStoreReadWriteTransaction transaction;
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+ public ShardReadWriteTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadWriteTransaction transaction, ActorRef shardActor, SchemaContext schemaContext) {
+ super(transactionChain, shardActor, schemaContext);
+ this.transaction = transaction;
+ }
+
+ public ShardReadWriteTransaction(DOMStoreReadWriteTransaction transaction, ActorRef shardActor, SchemaContext schemaContext) {
+ super( shardActor, schemaContext);
+ this.transaction = transaction;
+ }
+
+ @Override
+ public void handleReceive(Object message) throws Exception {
+ if (ReadData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ readData(transaction,ReadData.fromSerializable(message));
+ }else if (WriteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ writeData(transaction, WriteData.fromSerializable(message, schemaContext));
+ } else if (MergeData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ mergeData(transaction, MergeData.fromSerializable(message, schemaContext));
+ } else if (DeleteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ deleteData(transaction,DeleteData.fromSerizalizable(message));
+ } else if (ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ readyTransaction(transaction,new ReadyTransaction());
+ }else {
+ super.handleReceive(message);
+ }
+ }
+
+ protected void closeTransaction(CloseTransaction message) {
+ transaction.close();
+ getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
+ getSelf().tell(PoisonPill.getInstance(), getSelf());
+ }
+}
package org.opendaylight.controller.cluster.datastore;
import akka.actor.ActorRef;
-import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.google.common.base.Optional;
import com.google.common.util.concurrent.ListenableFuture;
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;
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.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.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 org.opendaylight.yangtools.yang.model.api.SchemaContext;
* <li> {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction}
* </p>
*/
-public class ShardTransaction extends AbstractUntypedActor {
-
- private final ActorRef shardActor;
- private 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 LoggingAdapter log =
- Logging.getLogger(getContext().system(), this);
-
- public ShardTransaction(DOMStoreReadWriteTransaction transaction,
- ActorRef shardActor, SchemaContext schemaContext) {
- this(null, transaction, shardActor, schemaContext);
- }
-
- public ShardTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadWriteTransaction transaction,
- ActorRef shardActor, SchemaContext schemaContext) {
- this.transactionChain = transactionChain;
- this.transaction = transaction;
- this.shardActor = shardActor;
- this.schemaContext = schemaContext;
- }
-
-
-
- public static Props props(final DOMStoreReadWriteTransaction transaction,
- final ActorRef shardActor, final SchemaContext schemaContext) {
- return Props.create(new Creator<ShardTransaction>() {
-
- @Override
- public ShardTransaction create() throws Exception {
- return new ShardTransaction(transaction, shardActor, schemaContext);
- }
- });
- }
-
- public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
- final ActorRef shardActor, final SchemaContext schemaContext) {
- return Props.create(new Creator<ShardTransaction>() {
-
- @Override
- public ShardTransaction create() throws Exception {
- return new ShardTransaction(transactionChain, transaction, shardActor, schemaContext);
- }
- });
+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 MutableCompositeModification modification =
+ new MutableCompositeModification();
+
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+
+ protected ShardTransaction(
+ ActorRef shardActor, SchemaContext schemaContext) {
+ this(null, shardActor, schemaContext);
+ }
+
+ protected ShardTransaction(DOMStoreTransactionChain transactionChain,
+ ActorRef shardActor, SchemaContext schemaContext) {
+ this.transactionChain = transactionChain;
+ //this.transaction = transaction;
+ this.shardActor = shardActor;
+ this.schemaContext = schemaContext;
+ }
+
+
+
+ public static Props props(final DOMStoreReadTransaction transaction,
+ final ActorRef shardActor, final SchemaContext schemaContext) {
+ return Props.create(new Creator<ShardTransaction>() {
+
+ @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<ShardTransaction>() {
+
+ @Override
+ public ShardTransaction create() throws Exception {
+ 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<ShardTransaction>() {
+
+ @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<ShardTransaction>() {
+
+ @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<ShardTransaction>() {
+
+ @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<ShardTransaction>() {
+
+ @Override
+ public ShardTransaction create() throws Exception {
+ return new ShardWriteTransaction(transactionChain, transaction, shardActor, schemaContext);
+ }
+ });
+ }
+
+
+ @Override
+ 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());
+ }else{
+ throw new Exception ("ShardTransaction:handleRecieve received an unknown message"+message);
}
-
-
- @Override
- public void handleReceive(Object message) throws Exception {
- if (ReadData.SERIALIZABLE_CLASS.equals(message.getClass())) {
- readData(ReadData.fromSerializable(message));
- } else if (WriteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
- writeData(WriteData.fromSerializable(message, schemaContext));
- } else if (MergeData.SERIALIZABLE_CLASS.equals(message.getClass())) {
- mergeData(MergeData.fromSerializable(message, schemaContext));
- } else if (DeleteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
- deleteData(DeleteData.fromSerizalizable(message));
- } else if (ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
- readyTransaction(new ReadyTransaction());
- } else 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());
- }else{
- throw new Exception ("Shard:handleRecieve received an unknown message"+message);
+ }
+
+ abstract protected void closeTransaction(CloseTransaction message);
+
+ protected void readData(DOMStoreReadTransaction transaction,ReadData message) {
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+ final YangInstanceIdentifier path = message.getPath();
+ final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
+ transaction.read(path);
+
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Optional<NormalizedNode<?, ?>> optional = future.get();
+ if (optional.isPresent()) {
+ sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
+ } else {
+ 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());
}
- }
- private void readData(ReadData message) {
- final ActorRef sender = getSender();
- final ActorRef self = getSelf();
- final YangInstanceIdentifier path = message.getPath();
- final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
- transaction.read(path);
-
- future.addListener(new Runnable() {
- @Override
- public void run() {
- try {
- Optional<NormalizedNode<?, ?>> optional = future.get();
- if (optional.isPresent()) {
- sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self);
- } else {
- 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());
- }
-
- }
- }, getContext().dispatcher());
- }
+ }
+ }, getContext().dispatcher());
+ }
- private void writeData(WriteData message) {
- modification.addModification(
- new WriteModification(message.getPath(), message.getData(),schemaContext));
- LOG.debug("writeData at path : " + message.getPath().toString());
- transaction.write(message.getPath(), message.getData());
- getSender().tell(new WriteDataReply().toSerializable(), 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());
+ transaction.write(message.getPath(), message.getData());
+ getSender().tell(new WriteDataReply().toSerializable(), getSelf());
+ }
- private void mergeData(MergeData message) {
- modification.addModification(
- new MergeModification(message.getPath(), message.getData(), schemaContext));
- LOG.debug("mergeData at path : " + message.getPath().toString());
- transaction.merge(message.getPath(), message.getData());
- getSender().tell(new MergeDataReply().toSerializable(), 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());
+ transaction.merge(message.getPath(), message.getData());
+ getSender().tell(new MergeDataReply().toSerializable(), getSelf());
+ }
- private void deleteData(DeleteData message) {
- modification.addModification(new DeleteModification(message.getPath()));
- transaction.delete(message.getPath());
- getSender().tell(new DeleteDataReply().toSerializable(), getSelf());
- }
+ protected void deleteData(DOMStoreWriteTransaction transaction, DeleteData message) {
+ modification.addModification(new DeleteModification(message.getPath()));
+ transaction.delete(message.getPath());
+ getSender().tell(new DeleteDataReply().toSerializable(), getSelf());
+ }
- private void readyTransaction(ReadyTransaction message) {
- DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
- ActorRef cohortActor = getContext().actorOf(
- ThreePhaseCommitCohort.props(cohort, shardActor, modification), "cohort");
- getSender()
- .tell(new ReadyTransactionReply(cohortActor.path()).toSerializable(), 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().toSerializable(), getSelf());
- getSelf().tell(PoisonPill.getInstance(), getSelf());
- }
+ }
- // These classes are in here for test purposes only
+ // These classes are in here for test purposes only
- static class GetCompositedModification {
+ static class GetCompositedModification {
- }
+ }
- static class GetCompositeModificationReply {
- private final CompositeModification modification;
+ static class GetCompositeModificationReply {
+ private final CompositeModification modification;
- GetCompositeModificationReply(CompositeModification modification) {
- this.modification = modification;
- }
+ GetCompositeModificationReply(CompositeModification modification) {
+ this.modification = modification;
+ }
- public CompositeModification getModification() {
- return modification;
- }
+ public CompositeModification getModification() {
+ return modification;
}
+ }
}
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;
}
}
+ private ActorRef createTypedTransactionActor(CreateTransaction createTransaction,String transactionId){
+ if(createTransaction.getTransactionType()== TransactionProxy.TransactionType.READ_ONLY.ordinal()){
+ return getContext().actorOf(
+ ShardTransaction.props( chain.newReadOnlyTransaction(), getSelf(), schemaContext), transactionId);
+
+ }else if (createTransaction.getTransactionType()== TransactionProxy.TransactionType.READ_WRITE.ordinal()){
+ return getContext().actorOf(
+ ShardTransaction.props( chain.newReadWriteTransaction(), getSelf(), schemaContext), transactionId);
+
+
+ }else if (createTransaction.getTransactionType()== TransactionProxy.TransactionType.WRITE_ONLY.ordinal()){
+ return getContext().actorOf(
+ ShardTransaction.props( chain.newWriteOnlyTransaction(), getSelf(), schemaContext), transactionId);
+ }else{
+ throw new IllegalArgumentException ("CreateTransaction message has unidentified transaction type="+createTransaction.getTransactionType()) ;
+ }
+ }
+
private void createTransaction(CreateTransaction createTransaction) {
- DOMStoreReadWriteTransaction transaction =
- chain.newReadWriteTransaction();
- ActorRef transactionActor = getContext().actorOf(ShardTransaction
- .props(chain, transaction, getContext().parent(), schemaContext), "shard-" + createTransaction.getTransactionId());
+
+ ActorRef transactionActor = createTypedTransactionActor(createTransaction, "shard-" + createTransaction.getTransactionId());
getSender()
.tell(new CreateTransactionReply(transactionActor.path().toString(),createTransaction.getTransactionId()).toSerializable(),
getSelf());
--- /dev/null
+/*
+ *
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+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.MergeData;
+import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.WriteData;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+
+/**
+ * @author: syedbahm
+ * Date: 8/6/14
+ */
+public class ShardWriteTransaction extends ShardTransaction {
+ private final DOMStoreWriteTransaction transaction;
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+ public ShardWriteTransaction(DOMStoreWriteTransaction transaction, ActorRef shardActor, SchemaContext schemaContext) {
+ super( shardActor, schemaContext);
+ this.transaction = transaction;
+
+ }
+
+ public ShardWriteTransaction(DOMStoreTransactionChain transactionChain, DOMStoreWriteTransaction transaction, ActorRef shardActor, SchemaContext schemaContext) {
+ super(transactionChain, shardActor, schemaContext);
+ this.transaction = transaction;
+ }
+
+ @Override
+ public void handleReceive(Object message) throws Exception {
+ if (WriteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ writeData(transaction, WriteData.fromSerializable(message, schemaContext));
+ } else if (MergeData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ mergeData(transaction, MergeData.fromSerializable(message, schemaContext));
+ } else if (DeleteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ deleteData(transaction,DeleteData.fromSerizalizable(message));
+ } else if (ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ readyTransaction(transaction,new ReadyTransaction());
+ }else {
+ super.handleReceive(message);
+ }
+ }
+
+ protected void closeTransaction(CloseTransaction message) {
+ transaction.close();
+ getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
+ getSelf().tell(PoisonPill.getInstance(), getSelf());
+ }
+}
try {
Object response = actorContext.executeShardOperation(shardName,
- new CreateTransaction(identifier).toSerializable(),
+ new CreateTransaction(identifier,this.transactionType.ordinal() ).toSerializable(),
ActorContext.ASK_DURATION);
if (response.getClass()
.equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
public class CreateTransaction implements SerializableMessage {
public static Class SERIALIZABLE_CLASS = ShardTransactionMessages.CreateTransaction.class;
private final String transactionId;
+ private final int transactionType;
- public CreateTransaction(String transactionId){
+ public CreateTransaction(String transactionId, int transactionType){
this.transactionId = transactionId;
+ this.transactionType = transactionType;
}
public String getTransactionId() {
return transactionId;
}
+ public int getTransactionType() { return transactionType;}
+
@Override
public Object toSerializable() {
- return ShardTransactionMessages.CreateTransaction.newBuilder().setTransactionId(transactionId).build();
+ return ShardTransactionMessages.CreateTransaction.newBuilder().setTransactionId(transactionId).setTransactionType(transactionType).build();
}
public static CreateTransaction fromSerializable(Object message){
- return new CreateTransaction(((ShardTransactionMessages.CreateTransaction)message).getTransactionId());
+ ShardTransactionMessages.CreateTransaction createTransaction = (ShardTransactionMessages.CreateTransaction)message;
+ return new CreateTransaction(createTransaction.getTransactionId(),createTransaction.getTransactionType() );
}
}
System.out.println("Successfully created transaction chain");
// 2. Create a Transaction on the TransactionChain
- transactionChain.tell(new CreateTransaction("txn-1").toSerializable(), getRef());
+ transactionChain.tell(new CreateTransaction("txn-1", TransactionProxy.TransactionType.WRITE_ONLY.ordinal() ).toSerializable(), getRef());
final ActorSelection transaction =
new ExpectMsg<ActorSelection>("CreateTransactionReply") {
new UpdateSchemaContext(TestModel.createTestContext()),
getRef());
- subject.tell(new CreateTransaction("txn-1").toSerializable(),
+ subject.tell(new CreateTransaction("txn-1", TransactionProxy.TransactionType.READ_ONLY.ordinal() ).toSerializable(),
getRef());
final String out = new ExpectMsg<String>("match hint") {
new Within(duration("1 seconds")) {
protected void run() {
- subject.tell(new CreateTransaction("txn-1").toSerializable(), getRef());
+ subject.tell(new CreateTransaction("txn-1", TransactionProxy.TransactionType.READ_ONLY.ordinal() ).toSerializable(), getRef());
final String out = new ExpectMsg<String>("match hint") {
// do not put code outside this method, will run afterwards
import akka.actor.Props;
import akka.actor.Terminated;
import akka.testkit.JavaTestKit;
+import akka.testkit.TestActorRef;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
+import org.junit.Assert;
import org.junit.Test;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
new JavaTestKit(getSystem()) {{
final ActorRef shard = getSystem().actorOf(Shard.props("config", Collections.EMPTY_MAP));
final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, testSchemaContext);
+ ShardTransaction.props(store.newReadOnlyTransaction(), shard, testSchemaContext);
final ActorRef subject = getSystem().actorOf(props, "testReadData");
new Within(duration("1 seconds")) {
new JavaTestKit(getSystem()) {{
final ActorRef shard = getSystem().actorOf(Shard.props("config", Collections.EMPTY_MAP));
final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, testSchemaContext);
+ ShardTransaction.props( store.newReadOnlyTransaction(), shard, testSchemaContext);
final ActorRef subject = getSystem().actorOf(props, "testReadDataWhenDataNotFound");
new Within(duration("1 seconds")) {
new JavaTestKit(getSystem()) {{
final ActorRef shard = getSystem().actorOf(Shard.props("config", Collections.EMPTY_MAP));
final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, TestModel.createTestContext());
+ ShardTransaction.props(store.newWriteOnlyTransaction(), shard, TestModel.createTestContext());
final ActorRef subject =
getSystem().actorOf(props, "testWriteData");
new JavaTestKit(getSystem()) {{
final ActorRef shard = getSystem().actorOf(Shard.props("config", Collections.EMPTY_MAP));
final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, TestModel.createTestContext());
+ ShardTransaction.props( store.newWriteOnlyTransaction(), shard, TestModel.createTestContext());
final ActorRef subject =
getSystem().actorOf(props, "testDeleteData");
new JavaTestKit(getSystem()) {{
final ActorRef shard = getSystem().actorOf(Shard.props("config", Collections.EMPTY_MAP));
final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, TestModel.createTestContext());
+ ShardTransaction.props( store.newReadWriteTransaction(), shard, TestModel.createTestContext());
final ActorRef subject =
getSystem().actorOf(props, "testReadyTransaction");
}};
}
+
+
+ @Test
+ public void testNegativePerformingWriteOperationOnReadTransaction() throws Exception {
+ try {
+
+ final ActorRef shard = getSystem().actorOf(Shard.props("config", Collections.EMPTY_MAP));
+ final Props props =
+ ShardTransaction.props(store.newReadOnlyTransaction(), shard, TestModel.createTestContext());
+ final TestActorRef subject = TestActorRef.apply(props,getSystem());
+
+ subject.receive(new DeleteData(TestModel.TEST_PATH).toSerializable(), ActorRef.noSender());
+ Assert.assertFalse(true);
+
+
+ } catch (Exception cs) {
+ assertEquals(cs.getClass().getSimpleName(), Exception.class.getSimpleName());
+ assertTrue(cs.getMessage().startsWith("ShardTransaction:handleRecieve received an unknown message"));
+ }
+ }
}