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%2FRemoteTransactionContext.java;h=ba64e29d7542ec2027c4d6f1cbabd11d54bd5af0;hb=b5cb353e3553a39f576c284119af75ffa5ea66a9;hp=67916cf1d2c4871fd759c8281c3908f2af5d31e8;hpb=0281535ab08fd795e42df66d25e9a904ff941ad7;p=controller.git

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 67916cf1d2..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;
@@ -49,7 +50,7 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
         this.actorContext = actorContext;
     }
 
-    private Future<Object> completeOperation(Future<Object> operationFuture){
+    private Future<Object> completeOperation(Future<Object> operationFuture) {
         operationFuture.onComplete(limiter, actorContext.getClientDispatcher());
         return operationFuture;
     }
@@ -62,8 +63,8 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
         return actorContext;
     }
 
-    protected Future<Object> executeOperationAsync(SerializableMessage msg) {
-        return completeOperation(actorContext.executeOperationAsync(getActor(), msg.toSerializable()));
+    protected Future<Object> executeOperationAsync(SerializableMessage msg, Timeout timeout) {
+        return completeOperation(actorContext.executeOperationAsync(getActor(), msg.toSerializable(), timeout));
     }
 
     @Override
@@ -74,11 +75,6 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
         actorContext.sendOperationAsync(getActor(), new CloseTransaction(getTransactionVersion()).toSerializable());
     }
 
-    @Override
-    public boolean supportsDirectCommit() {
-        return true;
-    }
-
     @Override
     public Future<Object> directCommit() {
         LOG.debug("Tx {} directCommit called", getIdentifier());
@@ -109,20 +105,19 @@ public class RemoteTransactionContext extends AbstractTransactionContext {
     }
 
     private BatchedModifications newBatchedModifications() {
-        return new BatchedModifications(getIdentifier().toString(), getTransactionVersion(),
-                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();
         }
     }
@@ -133,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;
@@ -160,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);
@@ -171,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.
@@ -185,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.asVersion(getTransactionVersion()));
+        Future<Object> future = executeOperationAsync(readCmd.asVersion(getTransactionVersion()),
+                actorContext.getOperationTimeout());
 
         future.onComplete(onComplete, actorContext.getClientDispatcher());
     }
@@ -210,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();
         }