Relax visibility on FrontendReadWriteTransaction methods

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / FrontendReadWriteTransaction.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java
index 2c2a287670f2e2676d42721c31641e181569a0e4..d80488ca2207ec1e2342cb44029ec90e1e9eaab6 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendReadWriteTransaction.java
@@ -32,8 +32,6 @@ import org.opendaylight.controller.cluster.access.commands.TransactionMerge;
 import org.opendaylight.controller.cluster.access.commands.TransactionModification;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
-import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
-import org.opendaylight.controller.cluster.access.commands.TransactionPurgeResponse;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
 import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
@@ -103,10 +101,8 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
             handleTransactionDoCommit((TransactionDoCommitRequest) request, envelope, now);
             return null;
         } else if (request instanceof TransactionAbortRequest) {
-            return handleTransactionAbort((TransactionAbortRequest) request, envelope, now);
-        } else if (request instanceof TransactionPurgeRequest) {
-            // No-op for now
-            return new TransactionPurgeResponse(request.getTarget(), request.getSequence());
+            handleTransactionAbort((TransactionAbortRequest) request, envelope, now);
+            return null;
         } else {
             throw new UnsupportedRequestException(request);
         }
@@ -145,11 +141,12 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
         });
     }
 
-    private TransactionSuccess<?> handleTransactionAbort(final TransactionAbortRequest request,
+    private void handleTransactionAbort(final TransactionAbortRequest request,
             final RequestEnvelope envelope, final long now) throws RequestException {
         if (readyCohort == null) {
-            openTransaction.abort();
-            return new TransactionAbortSuccess(getIdentifier(), request.getSequence());
+            openTransaction.abort(() -> recordAndSendSuccess(envelope, now,
+                new TransactionAbortSuccess(getIdentifier(), request.getSequence())));
+            return;
         }
 
         readyCohort.abort(new FutureCallback<Void>() {
@@ -168,7 +165,6 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
                 recordAndSendFailure(envelope, now, new RuntimeRequestException("Abort failed", failure));
             }
         });
-        return null;
     }
 
     private void coordinatedCommit(final RequestEnvelope envelope, final long now) {
@@ -202,7 +198,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
         });
     }
 
-    private void successfulDirectCanCommit(final RequestEnvelope envelope, final long startTime) {
+    void successfulDirectCanCommit(final RequestEnvelope envelope, final long startTime) {
         readyCohort.preCommit(new FutureCallback<DataTreeCandidate>() {
             @Override
             public void onSuccess(final DataTreeCandidate result) {
@@ -217,9 +213,8 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
         });
     }
 
-    private void successfulDirectPreCommit(final RequestEnvelope envelope, final long startTime) {
+    void successfulDirectPreCommit(final RequestEnvelope envelope, final long startTime) {
         readyCohort.commit(new FutureCallback<UnsignedLong>() {
-
             @Override
             public void onSuccess(final UnsignedLong result) {
                 successfulCommit(envelope, startTime);
@@ -233,7 +228,7 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
         });
     }
 
-    private void successfulCommit(final RequestEnvelope envelope, final long startTime) {
+    void successfulCommit(final RequestEnvelope envelope, final long startTime) {
         recordAndSendSuccess(envelope, startTime, new TransactionCommitSuccess(readyCohort.getIdentifier(),
             envelope.getMessage().getSequence()));
         readyCohort = null;
@@ -241,16 +236,22 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
 
     private void handleCommitLocalTransaction(final CommitLocalTransactionRequest request,
             final RequestEnvelope envelope, final long now) throws RequestException {
-        if (sealedModification.equals(request.getModification())) {
+        if (!sealedModification.equals(request.getModification())) {
+            LOG.warn("Expecting modification {}, commit request has {}", sealedModification, request.getModification());
+            throw new UnsupportedRequestException(request);
+        }
+
+        final java.util.Optional<Exception> optFailure = request.getDelayedFailure();
+        if (optFailure.isPresent()) {
+            readyCohort = history().createFailedCohort(getIdentifier(), sealedModification, optFailure.get());
+        } else {
             readyCohort = history().createReadyCohort(getIdentifier(), sealedModification);
+        }
 
-            if (request.isCoordinated()) {
-                coordinatedCommit(envelope, now);
-            } else {
-                directCommit(envelope, now);
-            }
+        if (request.isCoordinated()) {
+            coordinatedCommit(envelope, now);
         } else {
-            throw new UnsupportedRequestException(request);
+            directCommit(envelope, now);
         }
     }
 
@@ -298,9 +299,9 @@ final class FrontendReadWriteTransaction extends FrontendTransaction {
 
         switch (maybeProto.get()) {
             case ABORT:
-                openTransaction.abort();
+                openTransaction.abort(() -> replyModifySuccess(request.getSequence()));
                 openTransaction = null;
-                return replyModifySuccess(request.getSequence());
+                return null;
             case READY:
                 ensureReady();
                 return replyModifySuccess(request.getSequence());