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.BatchedModifications;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
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;
}
private Future<Object> completeOperation(Future<Object> operationFuture){
- operationFuture.onComplete(getLimiter(), actorContext.getClientDispatcher());
+ operationFuture.onComplete(limiter, actorContext.getClientDispatcher());
return operationFuture;
}
future.onComplete(onComplete, actorContext.getClientDispatcher());
}
+
+ /**
+ * 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();
+ }
+ }
+
+ @Override
+ public boolean usesOperationLimiting() {
+ return true;
+ }
}