X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FRemoteTransactionContext.java;h=af0c8714092ea37e82fb389b828cf415ff8bdf31;hp=7e8a2a00ebe3fb701cb66f0991ba1a0fee9e2cf0;hb=74926cb05f2e5e4937658ca61444f7d7c846eb00;hpb=d71b6614d6cdb5a98f086edeb56f5c52f365c61c diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContext.java index 7e8a2a00eb..af0c871409 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContext.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContext.java @@ -10,23 +10,17 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; import akka.dispatch.OnComplete; -import com.google.common.base.Optional; +import com.google.common.base.Preconditions; import com.google.common.util.concurrent.SettableFuture; +import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier; +import org.opendaylight.controller.cluster.datastore.messages.AbstractRead; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; -import org.opendaylight.controller.cluster.datastore.messages.DataExists; -import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply; -import org.opendaylight.controller.cluster.datastore.messages.ReadData; -import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; import org.opendaylight.controller.cluster.datastore.messages.SerializableMessage; -import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; -import org.opendaylight.controller.cluster.datastore.modification.MergeModification; +import org.opendaylight.controller.cluster.datastore.modification.AbstractModification; import org.opendaylight.controller.cluster.datastore.modification.Modification; -import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; -import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; -import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.Future; @@ -44,14 +38,16 @@ public class RemoteTransactionContext extends AbstractTransactionContext { private final ActorSelection actor; private final boolean isTxActorLocal; private final short remoteTransactionVersion; + private final OperationLimiter limiter; private BatchedModifications batchedModifications; private int totalBatchedModificationsSent; - protected RemoteTransactionContext(ActorSelection actor, + protected RemoteTransactionContext(TransactionIdentifier identifier, ActorSelection actor, ActorContext actorContext, boolean isTxActorLocal, short remoteTransactionVersion, OperationLimiter limiter) { - super(limiter); + super(identifier); + this.limiter = Preconditions.checkNotNull(limiter); this.actor = actor; this.actorContext = actorContext; this.isTxActorLocal = isTxActorLocal; @@ -59,7 +55,7 @@ public class RemoteTransactionContext extends AbstractTransactionContext { } private Future completeOperation(Future operationFuture){ - operationFuture.onComplete(getLimiter(), actorContext.getClientDispatcher()); + operationFuture.onComplete(limiter, actorContext.getClientDispatcher()); return operationFuture; } @@ -171,34 +167,22 @@ public class RemoteTransactionContext extends AbstractTransactionContext { } @Override - public void deleteData(YangInstanceIdentifier path) { - LOG.debug("Tx {} deleteData called path = {}", getIdentifier(), path); - - acquireOperation(); - batchModification(new DeleteModification(path)); - } - - @Override - public void mergeData(YangInstanceIdentifier path, NormalizedNode data) { - LOG.debug("Tx {} mergeData called path = {}", getIdentifier(), path); - - acquireOperation(); - batchModification(new MergeModification(path, data)); - } - - @Override - public void writeData(YangInstanceIdentifier path, NormalizedNode data) { - LOG.debug("Tx {} writeData called path = {}", getIdentifier(), path); + public void executeModification(AbstractModification modification) { + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {} executeModification {} called path = {}", getIdentifier(), modification.getClass() + .getSimpleName(), modification.getPath()); + } acquireOperation(); - batchModification(new WriteModification(path, data)); + batchModification(modification); } @Override - public void readData(final YangInstanceIdentifier path, - final SettableFuture>> returnFuture ) { - - LOG.debug("Tx {} readData called path = {}", getIdentifier(), path); + public void executeRead(final AbstractRead readCmd, final SettableFuture returnFuture) { + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {} executeRead {} called path = {}", getIdentifier(), readCmd.getClass().getSimpleName(), + readCmd.getPath()); + } // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the // public API contract. @@ -208,73 +192,40 @@ public class RemoteTransactionContext extends AbstractTransactionContext { OnComplete onComplete = new OnComplete() { @Override - public void onComplete(Throwable failure, Object readResponse) throws Throwable { + public void onComplete(Throwable failure, Object response) throws Throwable { if(failure != null) { - LOG.debug("Tx {} read operation failed: {}", getIdentifier(), failure); - returnFuture.setException(new ReadFailedException( - "Error reading data for path " + path, failure)); - + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {} {} operation failed: {}", getIdentifier(), readCmd.getClass().getSimpleName(), + failure); + } + returnFuture.setException(new ReadFailedException("Error checking " + readCmd.getClass().getSimpleName() + + " for path " + readCmd.getPath(), failure)); } else { - LOG.debug("Tx {} read operation succeeded", getIdentifier(), failure); - - if (readResponse instanceof ReadDataReply) { - ReadDataReply reply = (ReadDataReply) readResponse; - returnFuture.set(Optional.>fromNullable(reply.getNormalizedNode())); - - } else if (ReadDataReply.isSerializedType(readResponse)) { - ReadDataReply reply = ReadDataReply.fromSerializable(readResponse); - returnFuture.set(Optional.>fromNullable(reply.getNormalizedNode())); - - } else { - returnFuture.setException(new ReadFailedException( - "Invalid response reading data for path " + path)); + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {} {} operation succeeded", getIdentifier(), readCmd.getClass().getSimpleName()); } + readCmd.processResponse(response, returnFuture); } } }; - Future readFuture = executeOperationAsync(new ReadData(path)); + Future future = executeOperationAsync(readCmd); - readFuture.onComplete(onComplete, actorContext.getClientDispatcher()); + future.onComplete(onComplete, actorContext.getClientDispatcher()); } - @Override - public void dataExists(final YangInstanceIdentifier path, final SettableFuture returnFuture) { - - LOG.debug("Tx {} dataExists called path = {}", getIdentifier(), path); - - // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the - // public API contract. - - acquireOperation(); - sendBatchedModifications(); - - OnComplete onComplete = new OnComplete() { - @Override - public void onComplete(Throwable failure, Object response) throws Throwable { - if(failure != null) { - LOG.debug("Tx {} dataExists operation failed: {}", getIdentifier(), failure); - returnFuture.setException(new ReadFailedException( - "Error checking data exists for path " + path, failure)); - } else { - LOG.debug("Tx {} dataExists operation succeeded", getIdentifier(), failure); - - if (response instanceof DataExistsReply) { - returnFuture.set(Boolean.valueOf(((DataExistsReply) response).exists())); - - } else if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) { - returnFuture.set(Boolean.valueOf(DataExistsReply.fromSerializable(response).exists())); - - } else { - returnFuture.setException(new ReadFailedException( - "Invalid response checking exists for path " + path)); - } - } - } - }; - - Future future = executeOperationAsync(new DataExists(path)); + /** + * Acquire operation from the limiter if the hand-off has completed. If + * the hand-off is still ongoing, this method does nothing. + */ + private final void acquireOperation() { + if (isOperationHandOffComplete()) { + limiter.acquire(); + } + } - future.onComplete(onComplete, actorContext.getClientDispatcher()); + @Override + public boolean usesOperationLimiting() { + return true; } }