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=59bb4bfd77892e1a915563b3d435150de27c4ca1;hb=2641bb342bb87f8c5b6f91a1918ebfb08f5ffe40;hp=bb676e3757e5038d49dcbfb56a745e8e27d11fa5;hpb=818258c2370c687de93edc887b32019d25c34095;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 bb676e3757..59bb4bfd77 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 @@ -57,25 +57,29 @@ import org.opendaylight.yangtools.yang.model.api.SchemaContext; */ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering { + protected static final boolean SERIALIZED_REPLY = true; + private final ActorRef shardActor; private final SchemaContext schemaContext; private final ShardStats shardStats; private final String transactionID; + private final int txnClientVersion; protected ShardTransaction(ActorRef shardActor, SchemaContext schemaContext, - ShardStats shardStats, String transactionID) { + ShardStats shardStats, String transactionID, int txnClientVersion) { super("shard-tx"); //actor name override used for metering. This does not change the "real" actor name this.shardActor = shardActor; this.schemaContext = schemaContext; this.shardStats = shardStats; this.transactionID = transactionID; + this.txnClientVersion = txnClientVersion; } public static Props props(DOMStoreTransaction transaction, ActorRef shardActor, SchemaContext schemaContext,DatastoreContext datastoreContext, ShardStats shardStats, - String transactionID) { + String transactionID, int txnClientVersion) { return Props.create(new ShardTransactionCreator(transaction, shardActor, schemaContext, - datastoreContext, shardStats, transactionID)); + datastoreContext, shardStats, transactionID, txnClientVersion)); } protected abstract DOMStoreTransaction getDOMStoreTransaction(); @@ -92,6 +96,10 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering return schemaContext; } + protected int getTxnClientVersion() { + return txnClientVersion; + } + @Override public void handleReceive(Object message) throws Exception { if (message.getClass().equals(CloseTransaction.SERIALIZABLE_CLASS)) { @@ -116,23 +124,24 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering getSelf().tell(PoisonPill.getInstance(), getSelf()); } - protected void readData(DOMStoreReadTransaction transaction,ReadData message) { + protected void readData(DOMStoreReadTransaction transaction, ReadData message, final boolean returnSerialized) { final ActorRef sender = getSender(); final ActorRef self = getSelf(); final YangInstanceIdentifier path = message.getPath(); final CheckedFuture>, ReadFailedException> future = transaction.read(path); + future.addListener(new Runnable() { @Override public void run() { try { Optional> optional = future.checkedGet(); - if (optional.isPresent()) { - sender.tell(new ReadDataReply(schemaContext,optional.get()).toSerializable(), self); - } else { - sender.tell(new ReadDataReply(schemaContext,null).toSerializable(), self); - } + ReadDataReply readDataReply = new ReadDataReply(schemaContext, optional.orNull()); + + sender.tell((returnSerialized ? readDataReply.toSerializable(): + readDataReply), self); + } catch (Exception e) { shardStats.incrementFailedReadTransactionsCount(); sender.tell(new akka.actor.Status.Failure(e), self); @@ -142,12 +151,15 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering }, getContext().dispatcher()); } - protected void dataExists(DOMStoreReadTransaction transaction, DataExists message) { + protected void dataExists(DOMStoreReadTransaction transaction, DataExists message, + final boolean returnSerialized) { final YangInstanceIdentifier path = message.getPath(); try { Boolean exists = transaction.exists(path).checkedGet(); - getSender().tell(new DataExistsReply(exists).toSerializable(), getSelf()); + DataExistsReply dataExistsReply = new DataExistsReply(exists); + getSender().tell(returnSerialized ? dataExistsReply.toSerializable() : + dataExistsReply, getSelf()); } catch (ReadFailedException e) { getSender().tell(new akka.actor.Status.Failure(e),getSelf()); } @@ -164,16 +176,18 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering final DatastoreContext datastoreContext; final ShardStats shardStats; final String transactionID; + final int txnClientVersion; ShardTransactionCreator(DOMStoreTransaction transaction, ActorRef shardActor, SchemaContext schemaContext, DatastoreContext datastoreContext, - ShardStats shardStats, String transactionID) { + ShardStats shardStats, String transactionID, int txnClientVersion) { this.transaction = transaction; this.shardActor = shardActor; this.shardStats = shardStats; this.schemaContext = schemaContext; this.datastoreContext = datastoreContext; this.transactionID = transactionID; + this.txnClientVersion = txnClientVersion; } @Override @@ -181,13 +195,13 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering ShardTransaction tx; if(transaction instanceof DOMStoreReadWriteTransaction) { tx = new ShardReadWriteTransaction((DOMStoreReadWriteTransaction)transaction, - shardActor, schemaContext, shardStats, transactionID); + shardActor, schemaContext, shardStats, transactionID, txnClientVersion); } else if(transaction instanceof DOMStoreReadTransaction) { tx = new ShardReadTransaction((DOMStoreReadTransaction)transaction, shardActor, - schemaContext, shardStats, transactionID); + schemaContext, shardStats, transactionID, txnClientVersion); } else { tx = new ShardWriteTransaction((DOMStoreWriteTransaction)transaction, - shardActor, schemaContext, shardStats, transactionID); + shardActor, schemaContext, shardStats, transactionID, txnClientVersion); } tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());