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=ba64e29d7542ec2027c4d6f1cbabd11d54bd5af0;hp=a2560bd369d353e46cfbc23aeaa780c1057f3b0c;hb=e78622411319748472b5d9edab14eb6dc92cf6b1;hpb=11e722671c6dc194761471038a4e3ca5f7fd8970

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 a2560bd369..ba64e29d75 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,9 +10,10 @@ package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorSelection;
 import akka.dispatch.OnComplete;
+import akka.util.Timeout;
 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.access.concepts.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;
@@ -36,25 +37,20 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
 
     private final ActorContext actorContext;
     private final ActorSelection actor;
-    private final boolean isTxActorLocal;
-    private final short remoteTransactionVersion;
     private final OperationLimiter limiter;
 
     private BatchedModifications batchedModifications;
     private int totalBatchedModificationsSent;
 
     protected RemoteTransactionContext(TransactionIdentifier identifier, ActorSelection actor,
-            ActorContext actorContext, boolean isTxActorLocal,
-            short remoteTransactionVersion, OperationLimiter limiter) {
-        super(identifier);
+            ActorContext actorContext, short remoteTransactionVersion, OperationLimiter limiter) {
+        super(identifier, remoteTransactionVersion);
         this.limiter = Preconditions.checkNotNull(limiter);
         this.actor = actor;
         this.actorContext = actorContext;
-        this.isTxActorLocal = isTxActorLocal;
-        this.remoteTransactionVersion = remoteTransactionVersion;
     }
 
-    private Future<Object> completeOperation(Future<Object> operationFuture){
+    private Future<Object> completeOperation(Future<Object> operationFuture) {
         operationFuture.onComplete(limiter, actorContext.getClientDispatcher());
         return operationFuture;
     }
@@ -67,12 +63,8 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
         return actorContext;
     }
 
-    protected short getRemoteTransactionVersion() {
-        return remoteTransactionVersion;
-    }
-
-    protected Future<Object> executeOperationAsync(SerializableMessage msg) {
-        return completeOperation(actorContext.executeOperationAsync(getActor(), isTxActorLocal ? msg : msg.toSerializable()));
+    protected Future<Object> executeOperationAsync(SerializableMessage msg, Timeout timeout) {
+        return completeOperation(actorContext.executeOperationAsync(getActor(), msg.toSerializable(), timeout));
     }
 
     @Override
@@ -80,12 +72,7 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
         LOG.debug("Tx {} closeTransaction called", getIdentifier());
         TransactionContextCleanup.untrack(this);
 
-        actorContext.sendOperationAsync(getActor(), new CloseTransaction(remoteTransactionVersion).toSerializable());
-    }
-
-    @Override
-    public boolean supportsDirectCommit() {
-        return true;
+        actorContext.sendOperationAsync(getActor(), new CloseTransaction(getTransactionVersion()).toSerializable());
     }
 
     @Override
@@ -118,19 +105,19 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
     }
 
     private BatchedModifications newBatchedModifications() {
-        return new BatchedModifications(getIdentifier().toString(), remoteTransactionVersion, getIdentifier().getChainId());
+        return new BatchedModifications(getIdentifier(), getTransactionVersion());
     }
 
     private void batchModification(Modification modification) {
         incrementModificationCount();
-        if(batchedModifications == null) {
+        if (batchedModifications == null) {
             batchedModifications = newBatchedModifications();
         }
 
         batchedModifications.addModification(modification);
 
-        if(batchedModifications.getModifications().size() >=
-                actorContext.getDatastoreContext().getShardBatchedModificationCount()) {
+        if (batchedModifications.getModifications().size()
+                >= actorContext.getDatastoreContext().getShardBatchedModificationCount()) {
             sendBatchedModifications();
         }
     }
@@ -141,26 +128,26 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
 
     protected Future<Object> sendBatchedModifications(boolean ready, boolean doCommitOnReady) {
         Future<Object> sent = null;
-        if(ready || (batchedModifications != null && !batchedModifications.getModifications().isEmpty())) {
-            if(batchedModifications == null) {
+        if (ready || batchedModifications != null && !batchedModifications.getModifications().isEmpty()) {
+            if (batchedModifications == null) {
                 batchedModifications = newBatchedModifications();
             }
 
-            if(LOG.isDebugEnabled()) {
-                LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(),
-                        batchedModifications.getModifications().size(), ready);
-            }
+            LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(),
+                    batchedModifications.getModifications().size(), ready);
 
             batchedModifications.setReady(ready);
             batchedModifications.setDoCommitOnReady(doCommitOnReady);
             batchedModifications.setTotalMessagesSent(++totalBatchedModificationsSent);
-            sent = executeOperationAsync(batchedModifications);
 
-            if(ready) {
+            final BatchedModifications toSend = batchedModifications;
+            if (ready) {
                 batchedModifications = null;
             } else {
                 batchedModifications = newBatchedModifications();
             }
+
+            sent = executeOperationAsync(toSend, actorContext.getTransactionCommitOperationTimeout());
         }
 
         return sent;
@@ -168,10 +155,8 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
 
     @Override
     public void executeModification(AbstractModification modification) {
-        if(LOG.isDebugEnabled()) {
-            LOG.debug("Tx {} executeModification {} called path = {}", getIdentifier(), modification.getClass()
-                    .getSimpleName(), modification.getPath());
-        }
+        LOG.debug("Tx {} executeModification {} called path = {}", getIdentifier(),
+                modification.getClass().getSimpleName(), modification.getPath());
 
         acquireOperation();
         batchModification(modification);
@@ -179,10 +164,8 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
 
     @Override
     public <T> void executeRead(final AbstractRead<T> readCmd, final SettableFuture<T> returnFuture) {
-        if(LOG.isDebugEnabled()) {
-            LOG.debug("Tx {} executeRead {} called path = {}", getIdentifier(), readCmd.getClass().getSimpleName(),
-                    readCmd.getPath());
-        }
+        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.
@@ -193,23 +176,21 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
         OnComplete<Object> onComplete = new OnComplete<Object>() {
             @Override
             public void onComplete(Throwable failure, Object response) throws Throwable {
-                if(failure != null) {
-                    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));
+                if (failure != null) {
+                    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 {
-                    if(LOG.isDebugEnabled()) {
-                        LOG.debug("Tx {} {} operation succeeded", getIdentifier(), readCmd.getClass().getSimpleName());
-                    }
+                    LOG.debug("Tx {} {} operation succeeded", getIdentifier(), readCmd.getClass().getSimpleName());
                     readCmd.processResponse(response, returnFuture);
                 }
             }
         };
 
-        Future<Object> future = executeOperationAsync(readCmd);
+        Future<Object> future = executeOperationAsync(readCmd.asVersion(getTransactionVersion()),
+                actorContext.getOperationTimeout());
 
         future.onComplete(onComplete, actorContext.getClientDispatcher());
     }
@@ -218,7 +199,7 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
      * 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() {
+    private void acquireOperation() {
         if (isOperationHandOffComplete()) {
             limiter.acquire();
         }