NormalizedNodeAggregator should also report empty

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

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java
index 46e1924dd51e44d49f4ccf4f0a213fda82d6e948..addf5649931e30b843355e66766519bea8371311 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java
@@ -7,12 +7,14 @@
  */
 package org.opendaylight.controller.cluster.databroker.actors.dds;
 
+import static com.google.common.base.Verify.verify;
+
 import com.google.common.util.concurrent.FluentFuture;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.function.Consumer;
-import javax.annotation.Nullable;
 import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
@@ -47,7 +49,7 @@ import org.opendaylight.yangtools.util.concurrent.FluentFutures;
 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.DataTreeModification;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -99,17 +101,17 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
 
     @Override
     void doDelete(final YangInstanceIdentifier path) {
-        appendModification(new TransactionDelete(path), Optional.empty());
+        appendModification(new TransactionDelete(path), OptionalLong.empty());
     }
 
     @Override
-    void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        appendModification(new TransactionMerge(path, data), Optional.empty());
+    void doMerge(final YangInstanceIdentifier path, final NormalizedNode data) {
+        appendModification(new TransactionMerge(path, data), OptionalLong.empty());
     }
 
     @Override
-    void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        appendModification(new TransactionWrite(path, data), Optional.empty());
+    void doWrite(final YangInstanceIdentifier path, final NormalizedNode data) {
+        appendModification(new TransactionWrite(path, data), OptionalLong.empty());
     }
 
     private <T> FluentFuture<T> sendReadRequest(final AbstractReadTransactionRequest<?> request,
@@ -135,8 +137,8 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
     }
 
     @Override
-    FluentFuture<Optional<NormalizedNode<?, ?>>> doRead(final YangInstanceIdentifier path) {
-        final SettableFuture<Optional<NormalizedNode<?, ?>>> future = SettableFuture.create();
+    FluentFuture<Optional<NormalizedNode>> doRead(final YangInstanceIdentifier path) {
+        final SettableFuture<Optional<NormalizedNode>> future = SettableFuture.create();
         return sendReadRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), path,
             isSnapshotOnly()), t -> completeRead(path, future, t), future);
     }
@@ -149,35 +151,35 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
     }
 
     private void ensureFlushedBuider() {
-        ensureFlushedBuider(Optional.empty());
+        ensureFlushedBuider(OptionalLong.empty());
     }
 
-    private void ensureFlushedBuider(final Optional<Long> enqueuedTicks) {
+    private void ensureFlushedBuider(final OptionalLong enqueuedTicks) {
         if (builderBusy) {
             flushBuilder(enqueuedTicks);
         }
     }
 
-    private void flushBuilder(final Optional<Long> enqueuedTicks) {
+    private void flushBuilder(final OptionalLong enqueuedTicks) {
         final ModifyTransactionRequest request = builder.build();
         builderBusy = false;
 
         sendModification(request, enqueuedTicks);
     }
 
-    private void sendModification(final TransactionRequest<?> request, final Optional<Long> enqueuedTicks) {
+    private void sendModification(final TransactionRequest<?> request, final OptionalLong enqueuedTicks) {
         if (enqueuedTicks.isPresent()) {
-            enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.get().longValue());
+            enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.getAsLong());
         } else {
             sendRequest(request, response -> completeModify(request, response));
         }
     }
 
     private void appendModification(final TransactionModification modification) {
-        appendModification(modification, Optional.empty());
+        appendModification(modification, OptionalLong.empty());
     }
 
-    private void appendModification(final TransactionModification modification, final Optional<Long> enqueuedTicks) {
+    private void appendModification(final TransactionModification modification, final OptionalLong enqueuedTicks) {
         if (operationFailure == null) {
             ensureInitializedBuilder();
 
@@ -237,8 +239,8 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
         recordFinishedRequest(response);
     }
 
-    private void completeRead(final YangInstanceIdentifier path,
-            final SettableFuture<Optional<NormalizedNode<?, ?>>> future, final Response<?, ?> response) {
+    private void completeRead(final YangInstanceIdentifier path, final SettableFuture<Optional<NormalizedNode>> future,
+            final Response<?, ?> response) {
         LOG.debug("Read request for {} completed with {}", path, response);
 
         if (response instanceof ReadTransactionSuccess) {
@@ -274,7 +276,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
     }
 
     @Override
-    boolean sealAndSend(final Optional<Long> enqueuedTicks) {
+    boolean sealAndSend(final OptionalLong enqueuedTicks) {
         if (sendReadyOnSeal) {
             ensureInitializedBuilder();
             builder.setReady();
@@ -347,7 +349,9 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
             // Persistence protocol implies we are sealed, propagate the marker, but hold off doing other actions
             // until we know what we are going to do.
             if (markSealed()) {
-                sealOnly();
+                if (!sealOnly()) {
+                    LOG.debug("Proxy {} has a successor, which should receive seal through a separate request", this);
+                }
             }
 
             final TransactionRequest<?> tmp;
@@ -407,16 +411,16 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
     private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
             final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
         final DataTreeModification mod = request.getModification();
-        final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+        final OptionalLong optTicks = OptionalLong.of(enqueuedTicks);
 
         mod.applyToCursor(new AbstractDataTreeModificationCursor() {
             @Override
-            public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
+            public void write(final PathArgument child, final NormalizedNode data) {
                 appendModification(new TransactionWrite(current().node(child), data), optTicks);
             }
 
             @Override
-            public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
+            public void merge(final PathArgument child, final NormalizedNode data) {
                 appendModification(new TransactionMerge(current().node(child), data), optTicks);
             }
 
@@ -430,10 +434,10 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
     }
 
     @Override
-    void handleReplayedRemoteRequest(final TransactionRequest<?> request,
-            @Nullable final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+    void handleReplayedRemoteRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+            final long enqueuedTicks) {
         final Consumer<Response<?, ?>> cb = callback != null ? callback : resp -> { /* NOOP */ };
-        final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+        final OptionalLong optTicks = OptionalLong.of(enqueuedTicks);
 
         if (request instanceof ModifyTransactionRequest) {
             handleReplayedModifyTransactionRequest(enqueuedTicks, cb, (ModifyTransactionRequest) request);
@@ -488,7 +492,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction {
             // Persistence protocol implies we are sealed, propagate the marker, but hold off doing other actions
             // until we know what we are going to do.
             if (markSealed()) {
-                sealOnly();
+                verify(sealOnly(), "Attempted to replay seal on %s", this);
             }
 
             final TransactionRequest<?> tmp;