BUG-5280: add basic concept of ClientSnapshot

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / LocalProxyTransaction.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java
index e9941179c7aa7c599435defda9b2f6a9165ff862..f75d443fc79f4acc9d2b1007ae941f6c1482b494 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java
@@ -9,7 +9,6 @@ package org.opendaylight.controller.cluster.databroker.actors.dds;
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;

-import com.google.common.base.Verify;

 import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.Futures;
 import java.util.function.Consumer;
 import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.Futures;
 import java.util.function.Consumer;


@@ -20,17 +19,9 @@ import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactio
 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;

-import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;

 import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess;
 import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess;

-import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
-import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
-import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
-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.TransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;

-import org.opendaylight.controller.cluster.access.commands.TransactionWrite;

 import org.opendaylight.controller.cluster.access.concepts.Response;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
 import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor;
 import org.opendaylight.controller.cluster.access.concepts.Response;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
 import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor;


@@ -38,9 +29,7 @@ import org.opendaylight.mdsal.common.api.ReadFailedException;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;

-import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeModification;

 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;

-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModificationCursor;

 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;


@@ -61,161 +50,56 @@ import org.slf4j.LoggerFactory;
  * @author Robert Varga
  */
 @NotThreadSafe
  * @author Robert Varga
  */
 @NotThreadSafe

-final class LocalProxyTransaction extends AbstractProxyTransaction {
+abstract class LocalProxyTransaction extends AbstractProxyTransaction {

     private static final Logger LOG = LoggerFactory.getLogger(LocalProxyTransaction.class);
 
     private final TransactionIdentifier identifier;
 
     private static final Logger LOG = LoggerFactory.getLogger(LocalProxyTransaction.class);
 
     private final TransactionIdentifier identifier;
 

-    private CursorAwareDataTreeModification modification;
-    private CursorAwareDataTreeModification sealedModification;
-
-    LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier,
-        final CursorAwareDataTreeModification modification) {
+    LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier) {

         super(parent);
         this.identifier = Preconditions.checkNotNull(identifier);
         super(parent);
         this.identifier = Preconditions.checkNotNull(identifier);

-        this.modification = Preconditions.checkNotNull(modification);

     }
 
     @Override
     }
 
     @Override

-    public TransactionIdentifier getIdentifier() {
+    public final TransactionIdentifier getIdentifier() {

         return identifier;
     }
 
         return identifier;
     }
 

-    @Override
-    void doDelete(final YangInstanceIdentifier path) {
-        modification.delete(path);
-    }
+    abstract DataTreeSnapshot readOnlyView();

 
 

-    @Override
-    void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        modification.merge(path, data);
-    }
+    abstract void applyModifyTransactionRequest(ModifyTransactionRequest request,
+            @Nullable Consumer<Response<?, ?>> callback);

 
     @Override
 
     @Override

-    void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        modification.write(path, data);
+    final CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path) {
+        return Futures.immediateCheckedFuture(readOnlyView().readNode(path).isPresent());

     }
 
     @Override
     }
 
     @Override

-    CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path) {
-        return Futures.immediateCheckedFuture(modification.readNode(path).isPresent());
+    final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path) {
+        return Futures.immediateCheckedFuture(readOnlyView().readNode(path));

     }
 
     @Override
     }
 
     @Override

-    CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path) {
-        return Futures.immediateCheckedFuture(modification.readNode(path));
-    }
-
-    private RuntimeException abortedException() {
-        return new IllegalStateException("Tracker " + identifier + " has been aborted");
-    }
-
-    private RuntimeException submittedException() {
-        return new IllegalStateException("Tracker " + identifier + " has been submitted");
-    }
-
-    @Override
-    void doAbort() {
+    final void doAbort() {

         sendAbort(new AbortLocalTransactionRequest(identifier, localActor()), response -> {
             LOG.debug("Transaction {} abort completed with {}", identifier, response);
         });
     }
 
         sendAbort(new AbortLocalTransactionRequest(identifier, localActor()), response -> {
             LOG.debug("Transaction {} abort completed with {}", identifier, response);
         });
     }
 

-    @Override
-    CommitLocalTransactionRequest commitRequest(final boolean coordinated) {
-        final CommitLocalTransactionRequest ret = new CommitLocalTransactionRequest(identifier, nextSequence(),
-            localActor(), modification, coordinated);
-        modification = new FailedDataTreeModification(this::submittedException);
-        return ret;
-    }
-
-    @Override
-    void doSeal() {
-        modification.ready();
-        sealedModification = modification;
-    }
-
-    @Override
-    void flushState(final AbstractProxyTransaction successor) {
-        sealedModification.applyToCursor(new AbstractDataTreeModificationCursor() {
-            @Override
-            public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
-                successor.write(current().node(child), data);
-            }
-
-            @Override
-            public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
-                successor.merge(current().node(child), data);
-            }
-
-            @Override
-            public void delete(final PathArgument child) {
-                successor.delete(current().node(child));
-            }
-        });
-    }
-
-    DataTreeSnapshot getSnapshot() {
-        Preconditions.checkState(sealedModification != null, "Proxy %s is not sealed yet", identifier);
-        return sealedModification;
-    }
-
-    private void applyModifyTransactionRequest(final ModifyTransactionRequest request,
-            final @Nullable Consumer<Response<?, ?>> callback) {
-        for (TransactionModification mod : request.getModifications()) {
-            if (mod instanceof TransactionWrite) {
-                modification.write(mod.getPath(), ((TransactionWrite)mod).getData());
-            } else if (mod instanceof TransactionMerge) {
-                modification.merge(mod.getPath(), ((TransactionMerge)mod).getData());
-            } else if (mod instanceof TransactionDelete) {
-                modification.delete(mod.getPath());
-            } else {
-                throw new IllegalArgumentException("Unsupported modification " + mod);
-            }
-        }
-
-        final java.util.Optional<PersistenceProtocol> maybeProtocol = request.getPersistenceProtocol();
-        if (maybeProtocol.isPresent()) {
-            seal();
-            Verify.verify(callback != null, "Request {} has null callback", request);
-
-            switch (maybeProtocol.get()) {
-                case ABORT:
-                    sendAbort(callback);
-                    break;
-                case SIMPLE:
-                    sendRequest(commitRequest(false), callback);
-                    break;
-                case THREE_PHASE:
-                    sendRequest(commitRequest(true), callback);
-                    break;
-                default:
-                    throw new IllegalArgumentException("Unhandled protocol " + maybeProtocol.get());
-            }
-        }
-    }
-

     @Override
     void handleForwardedRemoteRequest(final TransactionRequest<?> request,
             final @Nullable Consumer<Response<?, ?>> callback) {
     @Override
     void handleForwardedRemoteRequest(final TransactionRequest<?> request,
             final @Nullable Consumer<Response<?, ?>> callback) {

-        LOG.debug("Applying forwarded request {}", request);
-

         if (request instanceof ModifyTransactionRequest) {
             applyModifyTransactionRequest((ModifyTransactionRequest) request, callback);
         } else if (request instanceof ReadTransactionRequest) {
             final YangInstanceIdentifier path = ((ReadTransactionRequest) request).getPath();
         if (request instanceof ModifyTransactionRequest) {
             applyModifyTransactionRequest((ModifyTransactionRequest) request, callback);
         } else if (request instanceof ReadTransactionRequest) {
             final YangInstanceIdentifier path = ((ReadTransactionRequest) request).getPath();

-            final Optional<NormalizedNode<?, ?>> result = modification.readNode(path);
+            final Optional<NormalizedNode<?, ?>> result = readOnlyView().readNode(path);

             callback.accept(new ReadTransactionSuccess(request.getTarget(), request.getSequence(), result));
         } else if (request instanceof ExistsTransactionRequest) {
             final YangInstanceIdentifier path = ((ExistsTransactionRequest) request).getPath();
             callback.accept(new ReadTransactionSuccess(request.getTarget(), request.getSequence(), result));
         } else if (request instanceof ExistsTransactionRequest) {
             final YangInstanceIdentifier path = ((ExistsTransactionRequest) request).getPath();

-            final boolean result = modification.readNode(path).isPresent();
+            final boolean result = readOnlyView().readNode(path).isPresent();

             callback.accept(new ExistsTransactionSuccess(request.getTarget(), request.getSequence(), result));
             callback.accept(new ExistsTransactionSuccess(request.getTarget(), request.getSequence(), result));

-        } else if (request instanceof TransactionPreCommitRequest) {
-            sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
-        } else if (request instanceof TransactionDoCommitRequest) {
-            sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
-        } else if (request instanceof TransactionAbortRequest) {
-            sendAbort(callback);

         } else {
             throw new IllegalArgumentException("Unhandled request " + request);
         }
         } else {
             throw new IllegalArgumentException("Unhandled request " + request);
         }


@@ -263,8 +147,6 @@ final class LocalProxyTransaction extends AbstractProxyTransaction {
             final Consumer<Response<?, ?>> callback) {
         if (request instanceof AbortLocalTransactionRequest) {
             successor.sendAbort(request, callback);
             final Consumer<Response<?, ?>> callback) {
         if (request instanceof AbortLocalTransactionRequest) {
             successor.sendAbort(request, callback);

-        } else if (request instanceof CommitLocalTransactionRequest) {
-            successor.sendCommit((CommitLocalTransactionRequest)request, callback);

         } else {
             throw new IllegalArgumentException("Unhandled request" + request);
         }
         } else {
             throw new IllegalArgumentException("Unhandled request" + request);
         }


@@ -272,18 +154,7 @@ final class LocalProxyTransaction extends AbstractProxyTransaction {
         LOG.debug("Forwarded request {} to successor {}", request, successor);
     }
 
         LOG.debug("Forwarded request {} to successor {}", request, successor);
     }
 

-    private void sendAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+    void sendAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {

         sendRequest(request, callback);
         sendRequest(request, callback);

-        modification = new FailedDataTreeModification(this::abortedException);
-    }
-
-    private void sendCommit(final CommitLocalTransactionRequest request, final Consumer<Response<?, ?>> callback) {
-        // Rebase old modification on new data tree.
-        try (DataTreeModificationCursor cursor = modification.createCursor(YangInstanceIdentifier.EMPTY)) {
-            request.getModification().applyToCursor(cursor);
-        }
-
-        seal();
-        sendRequest(commitRequest(request.isCoordinated()), callback);

     }
 }
     }
 }