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.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
+import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
+import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
import org.opendaylight.controller.cluster.datastore.messages.MergeData;
+import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
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.cluster.datastore.messages.WriteDataReply;
+import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
+import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
+import org.opendaylight.controller.cluster.datastore.modification.ImmutableCompositeModification;
+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.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
* 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.fromSerializable(message));
- } else if (ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
- readyTransaction(transaction,new ReadyTransaction());
- }else {
- super.handleReceive(message);
+
+ private final MutableCompositeModification modification = new MutableCompositeModification();
+ private final DOMStoreWriteTransaction transaction;
+
+ public ShardWriteTransaction(DOMStoreWriteTransaction transaction, ActorRef shardActor,
+ SchemaContext schemaContext, ShardStats shardStats, String transactionID) {
+ super(shardActor, schemaContext, shardStats, transactionID);
+ this.transaction = transaction;
+ }
+
+ @Override
+ protected DOMStoreTransaction getDOMStoreTransaction() {
+ return transaction;
+ }
+
+ @Override
+ public void handleReceive(Object message) throws Exception {
+ if(WriteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ writeData(transaction, WriteData.fromSerializable(message, getSchemaContext()));
+ } else if(MergeData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ mergeData(transaction, MergeData.fromSerializable(message, getSchemaContext()));
+ } else if(DeleteData.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ deleteData(transaction, DeleteData.fromSerializable(message));
+ } else if(ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) {
+ readyTransaction(transaction, new ReadyTransaction());
+ } else if (message instanceof GetCompositedModification) {
+ // This is here for testing only
+ getSender().tell(new GetCompositeModificationReply(
+ new ImmutableCompositeModification(modification)), getSelf());
+ } else {
+ super.handleReceive(message);
+ }
+ }
+
+ private void writeData(DOMStoreWriteTransaction transaction, WriteData message) {
+ modification.addModification(
+ new WriteModification(message.getPath(), message.getData(), getSchemaContext()));
+ if(LOG.isDebugEnabled()) {
+ 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(DOMStoreWriteTransaction transaction, MergeData message) {
+ modification.addModification(
+ new MergeModification(message.getPath(), message.getData(), getSchemaContext()));
+ if(LOG.isDebugEnabled()) {
+ 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(DOMStoreWriteTransaction transaction, DeleteData message) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("deleteData at path : " + message.getPath().toString());
+ }
+ modification.addModification(new DeleteModification(message.getPath()));
+ 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(DOMStoreWriteTransaction transaction, ReadyTransaction message) {
+ DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
+
+ getShardActor().forward(new ForwardedReadyTransaction(getTransactionID(), cohort, modification),
+ getContext());
}
- }
-
- protected void closeTransaction(CloseTransaction message) {
- transaction.close();
- getSender().tell(new CloseTransactionReply().toSerializable(), getSelf());
- getSelf().tell(PoisonPill.getInstance(), getSelf());
- }
-
- /**
- * The following method is used in unit testing only
- * hence the default scope.
- * This is done to test out failure cases.
- */
- public void forUnitTestOnlyExplicitTransactionClose() {
- transaction.close();
+
+ // These classes are in here for test purposes only
+
+ static class GetCompositedModification {
+ }
+
+ static class GetCompositeModificationReply {
+ private final CompositeModification modification;
+
+
+ GetCompositeModificationReply(CompositeModification modification) {
+ this.modification = modification;
+ }
+
+ public CompositeModification getModification() {
+ return modification;
+ }
}
}
Code Review / controller.git / blobdiff