BUG-8733: refactor IdInts listeners

Before doing any heavy work, this patch removes code duplication
between the two classes.

Change-Id: Ia17bf9fa31247f881a112dbb71c536e4ec7513ba
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/MdsalLowLevelTestProvider.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/MdsalLowLevelTestProvider.java
index e0e8d99d1aab3eac8df847bf7075de6e15b0257e..02ee427bff7b572ecbede9d36f9e9ad9a326a1b1 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/MdsalLowLevelTestProvider.java
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/MdsalLowLevelTestProvider.java
@@ -9,6 +9,7 @@
 package org.opendaylight.controller.clustering.it.provider;
 
 import static akka.actor.ActorRef.noSender;
+import static org.opendaylight.yangtools.yang.common.RpcResultBuilder.newError;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
@@ -20,16 +21,16 @@ import com.google.common.base.Optional;
 import com.google.common.base.Strings;
 import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ExecutionException;
+import java.util.Objects;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import org.opendaylight.controller.cluster.ActorSystemProvider;
 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientLocalHistory;
 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
@@ -40,6 +41,7 @@ import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory;
+import org.opendaylight.controller.clustering.it.provider.impl.DataListenerState;
 import org.opendaylight.controller.clustering.it.provider.impl.FlappingSingletonService;
 import org.opendaylight.controller.clustering.it.provider.impl.GetConstantService;
 import org.opendaylight.controller.clustering.it.provider.impl.IdIntsDOMDataTreeLIstener;
@@ -120,6 +122,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     private static final org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType CONTROLLER_CONFIG =
             org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION;
 
+    private static final ListenableFuture<RpcResult<Void>> VOID_SUCCESS = success(null);
+
     private final RpcProviderRegistry rpcRegistry;
     private final BindingAwareBroker.RpcRegistration<OdlMdsalLowlevelControlService> registration;
     private final DistributedShardFactory distributedShardFactory;
@@ -195,24 +199,21 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         if (getSingletonConstantRegistration == null) {
             LOG.debug("No get-singleton-constant registration present.");
-            final RpcError rpcError = RpcResultBuilder
-                    .newError(ErrorType.APPLICATION, "missing-registration", "No get-singleton-constant rpc registration present.");
-            final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
-            return Futures.immediateFuture(result);
+            return failure(newError(ErrorType.APPLICATION, "missing-registration",
+                "No get-singleton-constant rpc registration present."));
         }
 
         try {
             getSingletonConstantRegistration.close();
-            getSingletonConstantRegistration = null;
-
-            return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
         } catch (final Exception e) {
             LOG.debug("There was a problem closing the singleton constant service", e);
-            final RpcError rpcError = RpcResultBuilder
-                    .newError(ErrorType.APPLICATION, "error-closing", "There was a problem closing get-singleton-constant");
-            final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
-            return Futures.immediateFuture(result);
+            return failure(newError(ErrorType.APPLICATION, "error-closing",
+                "There was a problem closing get-singleton-constant"));
+        } finally {
+            getSingletonConstantRegistration = null;
         }
+
+        return VOID_SUCCESS;
     }
 
     @Override
@@ -226,16 +227,14 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         task.start();
 
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
     public Future<RpcResult<Void>> subscribeDtcl() {
-
         if (dtclReg != null) {
-            final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
-                    "There is already dataTreeChangeListener registered on id-ints list.");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Registration present.",
+                    "There is already dataTreeChangeListener registered on id-ints list."));
         }
 
         idIntsListener = new IdIntsListener();
@@ -246,7 +245,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
                                 CONTROLLER_CONFIG, WriteTransactionsHandler.ID_INT_YID),
                         idIntsListener);
 
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        LOG.debug("ClusteredDOMDataTreeChangeListener registered");
+        return VOID_SUCCESS;
     }
 
     @Override
@@ -257,29 +257,29 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
     @Override
     public Future<RpcResult<IsClientAbortedOutput>> isClientAborted() {
+        // FIXME: implement this
         return null;
     }
 
     @Override
     public Future<RpcResult<Void>> removeShardReplica(final RemoveShardReplicaInput input) {
+        // FIXME: implement this
         return null;
     }
 
     @Override
     public Future<RpcResult<Void>> subscribeYnl(final SubscribeYnlInput input) {
-
         LOG.debug("subscribe-ynl, input: {}", input);
 
         if (ynlRegistrations.containsKey(input.getId())) {
-            final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
-                    "There is already ynl listener registered for this id: " + input.getId());
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Registration present.",
+                    "There is already ynl listener registered for this id: " + input.getId()));
         }
 
         ynlRegistrations.put(input.getId(),
                 notificationService.registerNotificationListener(new YnlListener(input.getId())));
 
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return success(null);
     }
 
     @Override
@@ -305,35 +305,31 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         if (registration == null) {
             LOG.debug("No get-contexted-constant registration for context: {}", input.getContext());
-            final RpcError rpcError = RpcResultBuilder
-                    .newError(ErrorType.APPLICATION, "missing-registration", "No get-constant rpc registration present.");
-            final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
-            return Futures.immediateFuture(result);
+            return failure(newError(ErrorType.APPLICATION, "missing-registration",
+                "No get-constant rpc registration present."));
         }
 
         registration.close();
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
     public Future<RpcResult<Void>> registerSingletonConstant(final RegisterSingletonConstantInput input) {
-
         LOG.debug("Received register-singleton-constant rpc, input: {}", input);
 
         if (input.getConstant() == null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Invalid input.", "Constant value is null");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Invalid input.", "Constant value is null"));
         }
 
         getSingletonConstantRegistration =
                 SingletonGetConstantService.registerNew(singletonService, domRpcService, input.getConstant());
 
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
     public Future<RpcResult<Void>> registerDefaultConstant(final RegisterDefaultConstantInput input) {
+        // FIXME: implement this
         return null;
     }
 
@@ -341,16 +337,14 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     public Future<RpcResult<Void>> unregisterConstant() {
 
         if (globalGetConstantRegistration == null) {
-            final RpcError rpcError = RpcResultBuilder
-                    .newError(ErrorType.APPLICATION, "missing-registration", "No get-constant rpc registration present.");
-            final RpcResult<Void> result = RpcResultBuilder.<Void>failed().withRpcError(rpcError).build();
-            return Futures.immediateFuture(result);
+            return failure(newError(ErrorType.APPLICATION, "missing-registration",
+                "No get-constant rpc registration present."));
         }
 
         globalGetConstantRegistration.close();
         globalGetConstantRegistration = null;
 
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
@@ -358,50 +352,42 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("unregister-flapping-singleton received.");
 
         if (flappingSingletonService == null) {
-            final RpcError rpcError = RpcResultBuilder
-                    .newError(ErrorType.APPLICATION, "missing-registration", "No flapping-singleton registration present.");
-            final RpcResult<UnregisterFlappingSingletonOutput> result =
-                    RpcResultBuilder.<UnregisterFlappingSingletonOutput>failed().withRpcError(rpcError).build();
-            return Futures.immediateFuture(result);
+            return failure(newError(ErrorType.APPLICATION, "missing-registration",
+                "No flapping-singleton registration present."));
         }
 
         final long flapCount = flappingSingletonService.setInactive();
         flappingSingletonService = null;
 
-        final UnregisterFlappingSingletonOutput output =
-                new UnregisterFlappingSingletonOutputBuilder().setFlapCount(flapCount).build();
-
-        return Futures.immediateFuture(RpcResultBuilder.success(output).build());
+        return success(new UnregisterFlappingSingletonOutputBuilder().setFlapCount(flapCount).build());
     }
 
     @Override
     public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
+        // FIXME: implement this
         return null;
     }
 
     @Override
     public Future<RpcResult<Void>> subscribeDdtl() {
-
         if (ddtlReg != null) {
-            final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
-                    "There is already dataTreeChangeListener registered on id-ints list.");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Registration present.",
+                    "There is already dataTreeChangeListener registered on id-ints list."));
         }
 
         idIntsDdtl = new IdIntsDOMDataTreeLIstener();
 
         try {
-            ddtlReg =
-                    domDataTreeService.registerListener(idIntsDdtl,
-                            Collections.singleton(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
-                                    ProduceTransactionsHandler.ID_INT_YID))
-                            , true, Collections.emptyList());
+            ddtlReg = domDataTreeService.registerListener(idIntsDdtl,
+                Collections.singleton(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
+                    ProduceTransactionsHandler.ID_INT_YID)), true, Collections.emptyList());
         } catch (DOMDataTreeLoopException e) {
             LOG.error("Failed to register DOMDataTreeListener.", e);
-
+            return failure(newError(ErrorType.APPLICATION, "register-failed", e.getMessage()));
         }
 
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        LOG.debug("DOMDataTreeListener registered");
+        return VOID_SUCCESS;
     }
 
     @Override
@@ -409,21 +395,16 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("register-bound-constant: {}", input);
 
         if (input.getContext() == null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Invalid input.", "Context value is null");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Invalid input.", "Context value is null"));
         }
 
         if (input.getConstant() == null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Invalid input.", "Constant value is null");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Invalid input.", "Constant value is null"));
         }
 
         if (routedRegistrations.containsKey(input.getContext())) {
-            final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
-                    "There is already a rpc registered for context: " + input.getContext());
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Registration present.",
+                    "There is already a rpc registered for context: " + input.getContext()));
         }
 
         final DOMRpcImplementationRegistration<RoutedGetConstantService> registration =
@@ -431,7 +412,7 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
                         input.getConstant(), input.getContext());
 
         routedRegistrations.put(input.getContext(), registration);
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
@@ -439,14 +420,12 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("Received register-flapping-singleton.");
 
         if (flappingSingletonService != null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Registration present.", "flapping-singleton already registered");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
+                "flapping-singleton already registered"));
         }
 
         flappingSingletonService = new FlappingSingletonService(singletonService);
-
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
@@ -454,55 +433,52 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("Received unsubscribe-dtcl");
 
         if (idIntsListener == null || dtclReg == null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Dtcl missing.", "No DataTreeChangeListener registered.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Dtcl missing.", "No DataTreeChangeListener registered."));
         }
 
-        try {
-            idIntsListener.tryFinishProcessing().get(120, TimeUnit.SECONDS);
-        } catch (InterruptedException | ExecutionException | TimeoutException e) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "resource-denied-transport", "Unable to finish notification processing in 120 seconds.",
-                    "clustering-it", "clustering-it", e);
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
-                    .withRpcError(error).build());
-        }
-
-        dtclReg.close();
+        final ListenableFuture<DataListenerState> future = idIntsListener.tryFinishProcessing(dtclReg);
         dtclReg = null;
 
-        if (!idIntsListener.hasTriggered()) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.APPLICATION, "No notification received.", "id-ints listener has not received" +
-                            "any notifications.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
-                    .withRpcError(error).build());
+        return Futures.withFallback(Futures.transform(future, this::dtclOutput),
+            t -> failure(newError(ErrorType.RPC, "resource-denied-transport", "Failed to finish processing",
+                "clustering-it", "clustering-it", t)));
+    }
+
+    private RpcResult<UnsubscribeDtclOutput> dtclOutput(final DataListenerState state) {
+        if (state.changeCount() == 0) {
+            return failed(newError(ErrorType.APPLICATION, "No notification received.",
+                "id-ints listener has not received any notifications."));
         }
 
         final DOMDataReadOnlyTransaction rTx = domDataBroker.newReadOnlyTransaction();
-        try {
-            final Optional<NormalizedNode<?, ?>> readResult =
-                    rTx.read(CONTROLLER_CONFIG, WriteTransactionsHandler.ID_INT_YID).checkedGet();
-
-            if (!readResult.isPresent()) {
-                final RpcError error = RpcResultBuilder.newError(
-                        ErrorType.APPLICATION, "Final read empty.", "No data read from id-ints list.");
-                return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
-                        .withRpcError(error).build());
-            }
-
-            return Futures.immediateFuture(
-                    RpcResultBuilder.success(new UnsubscribeDtclOutputBuilder()
-                            .setCopyMatches(idIntsListener.checkEqual(readResult.get()))).build());
+        final Optional<NormalizedNode<?, ?>> readResult;
 
+        try {
+            readResult = rTx.read(CONTROLLER_CONFIG, WriteTransactionsHandler.ID_INT_YID).checkedGet();
         } catch (final ReadFailedException e) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.APPLICATION, "Read failed.", "Final read from id-ints failed.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDtclOutput>failed()
-                    .withRpcError(error).build());
+            return failed(newError(ErrorType.APPLICATION, "Read failed.", "Final read from id-ints failed."));
+        } finally {
+            rTx.close();
+        }
 
+        final NormalizedNode<?, ?> expected = state.lastData().orNull();
+        final NormalizedNode<?, ?> actual = readResult.orNull();
+        final boolean equal = Objects.equals(expected, actual);
+        if (!equal) {
+            LOG.debug("Expected result {} read resulted in {}", expected, actual);
         }
+
+        final RpcResultBuilder<UnsubscribeDtclOutput> b = RpcResultBuilder.success(new UnsubscribeDtclOutputBuilder()
+            .setCopyMatches(equal).build());
+
+//        for (DataListenerViolation violation : state.violations()) {
+//            final Optional<NormalizedNodeDiff> diff = violation.toDiff();
+//            if (diff.isPresent()) {
+//                b.withWarning(ErrorType.APPLICATION, "Sequence mismatch", diff.get().toString());
+//            }
+//        }
+
+        return b.build();
     }
 
     @Override
@@ -514,6 +490,7 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
     @Override
     public Future<RpcResult<Void>> deconfigureIdIntsShard() {
+        // FIXME: implement this
         return null;
     }
 
@@ -522,11 +499,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("Received unsubscribe-ynl, input: {}", input);
 
         if (!ynlRegistrations.containsKey(input.getId())) {
-            final RpcError rpcError = RpcResultBuilder
-                    .newError(ErrorType.APPLICATION, "missing-registration", "No ynl listener with this id registered.");
-            final RpcResult<UnsubscribeYnlOutput> result =
-                    RpcResultBuilder.<UnsubscribeYnlOutput>failed().withRpcError(rpcError).build();
-            return Futures.immediateFuture(result);
+            return failure(newError(ErrorType.APPLICATION, "missing-registration",
+                "No ynl listener with this id registered."));
         }
 
         final ListenerRegistration<YnlListener> registration = ynlRegistrations.remove(input.getId());
@@ -534,7 +508,7 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         registration.close();
 
-        return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeYnlOutput>success().withResult(output).build());
+        return success(output);
     }
 
     @Override
@@ -542,10 +516,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
             final CheckPublishNotificationsInput input) {
 
         final PublishNotificationsTask task = publishNotificationsTasks.get(input.getId());
-
         if (task == null) {
-            return Futures.immediateFuture(RpcResultBuilder.success(
-                    new CheckPublishNotificationsOutputBuilder().setActive(false)).build());
+            return success(new CheckPublishNotificationsOutputBuilder().setActive(Boolean.FALSE).build());
         }
 
         final CheckPublishNotificationsOutputBuilder checkPublishNotificationsOutputBuilder =
@@ -558,10 +530,7 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
             checkPublishNotificationsOutputBuilder.setLastError(task.getLastError().toString() + sw.toString());
         }
 
-        final CheckPublishNotificationsOutput output =
-                checkPublishNotificationsOutputBuilder.setPublishCount(task.getCurrentNotif()).build();
-
-        return Futures.immediateFuture(RpcResultBuilder.success(output).build());
+        return success(checkPublishNotificationsOutputBuilder.setPublishCount(task.getCurrentNotif()).build());
     }
 
     @Override
@@ -576,9 +545,7 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         final String shardName = input.getShardName();
         if (Strings.isNullOrEmpty(shardName)) {
-            final RpcError rpcError = RpcResultBuilder.newError(ErrorType.APPLICATION, "bad-element",
-                    "A valid shard name must be specified");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(rpcError).build());
+            return failure(newError(ErrorType.APPLICATION, "bad-element", "A valid shard name must be specified"));
         }
 
         return shutdownShardGracefully(shardName);
@@ -589,11 +556,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("Received shutdown-prefix-shard-replica rpc, input: {}", input);
 
         final InstanceIdentifier<?> shardPrefix = input.getPrefix();
-
         if (shardPrefix == null) {
-            final RpcError rpcError = RpcResultBuilder.newError(ErrorType.APPLICATION, "bad-element",
-                    "A valid shard prefix must be specified");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(rpcError).build());
+            return failure(newError(ErrorType.APPLICATION, "bad-element", "A valid shard prefix must be specified"));
         }
 
         final YangInstanceIdentifier shardPath = bindingNormalizedNodeSerializer.toYangInstanceIdentifier(shardPrefix);
@@ -624,11 +588,10 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         shutdownShardAsk.future().onComplete(new OnComplete<Boolean>() {
             @Override
-            public void onComplete(final Throwable throwable, final Boolean gracefulStopResult) throws Throwable {
+            public void onComplete(final Throwable throwable, final Boolean gracefulStopResult) {
                 if (throwable != null) {
-                    final RpcResult<Void> failedResult = RpcResultBuilder.<Void>failed()
-                            .withError(ErrorType.APPLICATION, "Failed to gracefully shutdown shard", throwable).build();
-                    rpcResult.set(failedResult);
+                    rpcResult.set(RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION,
+                        "Failed to gracefully shutdown shard", throwable).build());
                 } else {
                     // according to Patterns.gracefulStop API, we don't have to
                     // check value of gracefulStopResult
@@ -645,23 +608,21 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("Received register-constant rpc, input: {}", input);
 
         if (input.getConstant() == null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Invalid input.", "Constant value is null");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Invalid input.", "Constant value is null"));
         }
 
         if (globalGetConstantRegistration != null) {
-            final RpcError error = RpcResultBuilder.newError(ErrorType.RPC, "Registration present.",
-                    "There is already a get-constant rpc registered.");
-            return Futures.immediateFuture(RpcResultBuilder.<Void>failed().withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Registration present.",
+                    "There is already a get-constant rpc registered."));
         }
 
         globalGetConstantRegistration = GetConstantService.registerNew(domRpcService, input.getConstant());
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        return VOID_SUCCESS;
     }
 
     @Override
     public Future<RpcResult<Void>> unregisterDefaultConstant() {
+        // FIXME: implement this
         return null;
     }
 
@@ -670,53 +631,40 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
         LOG.debug("Received unsubscribe-ddtl.");
 
         if (idIntsDdtl == null || ddtlReg == null) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "Ddtl missing.", "No DOMDataTreeListener registered.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed().withRpcError(error).build());
-        }
-
-        try {
-            idIntsDdtl.tryFinishProcessing().get(120, TimeUnit.SECONDS);
-        } catch (InterruptedException | ExecutionException | TimeoutException e) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.RPC, "resource-denied-transport", "Unable to finish notification processing in 120 seconds.",
-                    "clustering-it", "clustering-it", e);
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
-                    .withRpcError(error).build());
+            return failure(newError(ErrorType.RPC, "Ddtl missing.", "No DOMDataTreeListener registered."));
         }
 
-        ddtlReg.close();
+        final ListenableFuture<DataListenerState> future = idIntsDdtl.tryFinishProcessing(ddtlReg);
         ddtlReg = null;
 
-        if (!idIntsDdtl.hasTriggered()) {
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.APPLICATION, "No notification received.", "id-ints listener has not received" +
-                            "any notifications.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
-                    .withRpcError(error).build());
+        return Futures.withFallback(Futures.transform(future, this::ddtlOutput),
+            t -> failure(newError(ErrorType.RPC, "resource-denied-transport", "Failed to finish processing",
+                "clustering-it", "clustering-it", t)));
+    }
+
+    private RpcResult<UnsubscribeDdtlOutput> ddtlOutput(final DataListenerState state) {
+        if (state.changeCount() == 0) {
+            return failed(newError(ErrorType.APPLICATION, "No notification received.",
+                "id-ints listener has not received any notifications."));
         }
 
         final String shardName = ClusterUtils.getCleanShardName(ProduceTransactionsHandler.ID_INTS_YID);
         LOG.debug("Creating distributed datastore client for shard {}", shardName);
 
         final ActorContext actorContext = configDataStore.getActorContext();
-        final Props distributedDataStoreClientProps =
-                SimpleDataStoreClientActor.props(actorContext.getCurrentMemberName(),
-                        "Shard-" + shardName, actorContext, shardName);
+        final Props distributedDataStoreClientProps = SimpleDataStoreClientActor.props(
+            actorContext.getCurrentMemberName(), "Shard-" + shardName, actorContext, shardName);
 
         final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
         final DataStoreClient distributedDataStoreClient;
         try {
-            distributedDataStoreClient = SimpleDataStoreClientActor
-                    .getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
+            distributedDataStoreClient = SimpleDataStoreClientActor.getDistributedDataStoreClient(clientActor, 30,
+                TimeUnit.SECONDS);
         } catch (final Exception e) {
             LOG.error("Failed to get actor for {}", distributedDataStoreClientProps, e);
             clientActor.tell(PoisonPill.getInstance(), noSender());
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.APPLICATION, "Unable to create ds client for read.",
-                    "Unable to create ds client for read.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
-                    .withRpcError(error).build());
+            return failed(newError(ErrorType.APPLICATION, "Unable to create ds client for read.",
+                    "Unable to create ds client for read."));
         }
 
         final ClientLocalHistory localHistory = distributedDataStoreClient.createLocalHistory();
@@ -727,29 +675,47 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
 
         tx.abort();
         localHistory.close();
-        try {
-            final Optional<NormalizedNode<?, ?>> optional = read.checkedGet();
-            if (!optional.isPresent()) {
-                LOG.warn("Final read from client is empty.");
-                final RpcError error = RpcResultBuilder.newError(
-                        ErrorType.APPLICATION, "Read failed.", "Final read from id-ints is empty.");
-                return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
-                        .withRpcError(error).build());
-            }
-
-            return Futures.immediateFuture(
-                    RpcResultBuilder.success(new UnsubscribeDdtlOutputBuilder()
-                            .setCopyMatches(idIntsDdtl.checkEqual(optional.get()))).build());
 
+        final Optional<NormalizedNode<?, ?>> readResult;
+        try {
+            readResult = read.checkedGet();
         } catch (org.opendaylight.mdsal.common.api.ReadFailedException e) {
             LOG.error("Unable to read data to verify ddtl data.", e);
-            final RpcError error = RpcResultBuilder.newError(
-                    ErrorType.APPLICATION, "Read failed.", "Final read from id-ints failed.");
-            return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeDdtlOutput>failed()
-                    .withRpcError(error).build());
+            return failed(newError( ErrorType.APPLICATION, "Read failed.", "Final read from id-ints failed."));
         } finally {
             distributedDataStoreClient.close();
             clientActor.tell(PoisonPill.getInstance(), noSender());
         }
+
+        // FIXME run a diff
+        final NormalizedNode<?, ?> expected = state.lastData().orNull();
+        final NormalizedNode<?, ?> actual = readResult.orNull();
+        final boolean equal = Objects.equals(expected, actual);
+        if (!equal) {
+            LOG.debug("Expected result {} read resulted in {}", expected, actual);
+        }
+        final RpcResultBuilder<UnsubscribeDdtlOutput> b = RpcResultBuilder.success(new UnsubscribeDdtlOutputBuilder()
+            .setCopyMatches(equal).build());
+
+//        for (DataListenerViolation violation : state.violations()) {
+//            final Optional<NormalizedNodeDiff> diff = violation.toDiff();
+//            if (diff.isPresent()) {
+//                b.withWarning(ErrorType.APPLICATION, "Sequence mismatch", diff.get().toString());
+//            }
+//        }
+
+        return b.build();
+    }
+
+    private static <T> RpcResult<T> failed(final RpcError error) {
+        return RpcResultBuilder.<T>failed().withRpcError(error).build();
+    }
+
+    private static <T> ListenableFuture<RpcResult<T>> failure(final RpcError error) {
+        return Futures.immediateFuture(failed(error));
+    }
+
+    private static <T> ListenableFuture<RpcResult<T>> success(final T result) {
+        return Futures.immediateFuture(RpcResultBuilder.success(result).build());
     }
 }

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/NormalizedNodeDiff.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/NormalizedNodeDiff.java
new file mode 100644 (file)
index 0000000..2cb97cc
--- /dev/null
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/NormalizedNodeDiff.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) 2017 Pantheon Technologies, s.r.o. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.clustering.it.provider;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Optional;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+public abstract class NormalizedNodeDiff {
+    private static final class Missing extends NormalizedNodeDiff {
+        private final NormalizedNode<?, ?> expected;
+
+        Missing(final NormalizedNode<?, ?> expected) {
+            this.expected = checkNotNull(expected);
+        }
+
+        @Override
+        public String toString() {
+            return MoreObjects.toStringHelper(this).add("expected", expected).toString();
+        }
+    }
+
+    private static final class Unexpected extends NormalizedNodeDiff {
+        private final NormalizedNode<?, ?> actual;
+
+        Unexpected(final NormalizedNode<?, ?> actual) {
+            this.actual = checkNotNull(actual);
+        }
+
+        @Override
+        public String toString() {
+            return MoreObjects.toStringHelper(this).add("actual", actual).toString();
+        }
+    }
+
+    private static final class Regular extends NormalizedNodeDiff {
+        private final NormalizedNode<?, ?> expected;
+        private final NormalizedNode<?, ?> actual;
+
+        Regular(final NormalizedNode<?, ?> expected, final NormalizedNode<?, ?> actual) {
+            this.expected = checkNotNull(expected);
+            this.actual = checkNotNull(actual);
+        }
+
+        @Override
+        public String toString() {
+            return MoreObjects.toStringHelper(this).add("expected", expected).add("actual", actual).toString();
+        }
+    }
+
+    public static Optional<NormalizedNodeDiff> compute(final NormalizedNode<?, ?> expected,
+            final NormalizedNode<?, ?> actual) {
+        if (expected == null) {
+            return actual == null ? Optional.absent() : Optional.of(new Unexpected(actual));
+        }
+        if (actual == null) {
+            return Optional.of(new Missing(expected));
+        }
+        if (actual.equals(expected)) {
+            return Optional.absent();
+        }
+
+        // TODO: better diff
+        return Optional.of(new Regular(expected, actual));
+    }
+}

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/AbstractDataListener.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/AbstractDataListener.java
new file mode 100644 (file)
index 0000000..c8ea1a9
--- /dev/null
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/AbstractDataListener.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2017 Pantheon Technologies, s.r.o. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.clustering.it.provider.impl;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.GuardedBy;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractDataListener {
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractDataListener.class);
+    private static final long MAX_ELAPSED_NANOS = TimeUnit.SECONDS.toNanos(4);
+
+    @GuardedBy("ticksSinceLast")
+    private final Stopwatch ticksSinceLast = Stopwatch.createUnstarted();
+
+    private DataListenerState state = DataListenerState.initial();
+    private ScheduledFuture<?> scheduledFuture;
+
+    AbstractDataListener() {
+        // Do not allow instantiation from outside of the class
+    }
+
+    public final ListenableFuture<DataListenerState> tryFinishProcessing(final ListenerRegistration<?> ddtlReg) {
+        final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+        final SettableFuture<DataListenerState> future = SettableFuture.create();
+
+        scheduledFuture = executorService.scheduleAtFixedRate(() -> {
+            final long elapsed;
+            synchronized (ticksSinceLast) {
+                elapsed = ticksSinceLast.elapsed(TimeUnit.NANOSECONDS);
+            }
+
+            if (elapsed > MAX_ELAPSED_NANOS) {
+                ddtlReg.close();
+                future.set(state);
+                scheduledFuture.cancel(false);
+                executorService.shutdown();
+            }
+        }, 0, 1, TimeUnit.SECONDS);
+        return future;
+    }
+
+    final void onReceivedChanges(@Nonnull final Collection<DataTreeCandidate> changes) {
+        // do not log the change into debug, only use trace since it will lead to OOM on default heap settings
+        LOG.debug("Received {} data tree changed events", changes.size());
+        changes.forEach(change -> {
+            LOG.trace("Processing change {}", change);
+            state = state.append(change);
+        });
+
+        synchronized (this) {
+            ticksSinceLast.reset().start();
+        }
+    }
+
+    final void onReceivedError(final Collection<? extends Throwable> errors) {
+        final Iterator<? extends Throwable> it = errors.iterator();
+        final Throwable first = it.next();
+        it.forEachRemaining(first::addSuppressed);
+
+        LOG.error("Listener failed", first);
+
+        // FIXME: mark the failure
+    }
+}

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/DataListenerState.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/DataListenerState.java
new file mode 100644 (file)
index 0000000..22999f9
--- /dev/null
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/DataListenerState.java
@@ -0,0 +1,152 @@
+/*
+ * Copyright (c) 2017 Pantheon Technologies, s.r.o. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.clustering.it.provider.impl;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
+import com.google.common.base.Optional;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
+
+public abstract class DataListenerState {
+    private static final class Initial extends DataListenerState {
+        Initial() {
+            super(new ArrayList<>(1));
+        }
+
+        @Override
+        public long changeCount() {
+            return 0;
+        }
+
+        @Override
+        public Optional<NormalizedNode<?, ?>> lastData() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        Optional<DataListenerViolation> validate(final long sequence, final ModificationType type,
+                final Optional<NormalizedNode<?, ?>> beforeOpt, final Optional<NormalizedNode<?, ?>> afterOpt) {
+            return beforeOpt.transform(before -> new DataListenerViolation(sequence, null, before));
+        }
+
+    }
+
+    private abstract static class Subsequent extends DataListenerState {
+        private final long changeCount;
+
+        Subsequent(final List<DataListenerViolation> violations, final long changeCount) {
+            super(violations);
+            this.changeCount = changeCount;
+        }
+
+        @Override
+        public final long changeCount() {
+            return changeCount;
+        }
+    }
+
+    private static final class Absent extends Subsequent {
+        Absent(final List<DataListenerViolation> violations, final long changeCount) {
+            super(violations, changeCount);
+        }
+
+        @Override
+        public Optional<NormalizedNode<?, ?>> lastData() {
+            return Optional.absent();
+        }
+
+        @Override
+        Optional<DataListenerViolation> validate(final long sequence, final ModificationType type,
+                final Optional<NormalizedNode<?, ?>> beforeOpt, final Optional<NormalizedNode<?, ?>> afterOpt) {
+            return beforeOpt.transform(before -> new DataListenerViolation(sequence, null, before));
+        }
+    }
+
+    private static final class Present extends Subsequent {
+        private final NormalizedNode<?, ?> data;
+
+        Present(final List<DataListenerViolation> violations, final long changeCount, final NormalizedNode<?, ?> data) {
+            super(violations, changeCount);
+            this.data = checkNotNull(data);
+        }
+
+        @Override
+        public Optional<NormalizedNode<?, ?>> lastData() {
+            return Optional.of(data);
+        }
+
+        @Override
+        Optional<DataListenerViolation> validate(final long sequence, final ModificationType type,
+                final Optional<NormalizedNode<?, ?>> beforeOpt, final Optional<NormalizedNode<?, ?>> afterOpt) {
+            if (!beforeOpt.isPresent()) {
+                return Optional.of(new DataListenerViolation(sequence, data, null));
+            }
+
+            final NormalizedNode<?, ?> before = beforeOpt.get();
+            // Identity check to keep things fast. We'll run a diff to eliminate false positives later
+            if (data != before) {
+                return Optional.of(new DataListenerViolation(sequence, data, before));
+            }
+
+            return Optional.absent();
+        }
+    }
+
+    private final List<DataListenerViolation> violations;
+
+    DataListenerState(final List<DataListenerViolation> violations) {
+        this.violations = checkNotNull(violations);
+    }
+
+    static DataListenerState initial() {
+        return new Initial();
+    }
+
+    final DataListenerState append(final DataTreeCandidate change) {
+        final DataTreeCandidateNode root = change.getRootNode();
+        final Optional<NormalizedNode<?, ?>> beforeOpt = root.getDataBefore();
+        final Optional<NormalizedNode<?, ?>> afterOpt = root.getDataAfter();
+        final long count = changeCount() + 1;
+
+        final Optional<DataListenerViolation> opt = validate(count, root.getModificationType(), beforeOpt, afterOpt);
+        if (opt.isPresent()) {
+            violations.add(opt.get());
+        }
+
+        return afterOpt.isPresent() ? new Present(violations, count, afterOpt.get()) : new Absent(violations, count);
+    }
+
+    public final List<DataListenerViolation> violations() {
+        return Collections.unmodifiableList(violations);
+    }
+
+
+    public abstract long changeCount();
+
+    public abstract Optional<NormalizedNode<?, ?>> lastData();
+
+    abstract Optional<DataListenerViolation> validate(final long sequence, final ModificationType type,
+            final Optional<NormalizedNode<?, ?>> beforeOpt, final Optional<NormalizedNode<?, ?>> afterOpt);
+
+    @Override
+    public final String toString() {
+        return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
+    }
+
+    ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
+        return toStringHelper.add("changeCount", changeCount());
+    }
+}

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/DataListenerViolation.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/DataListenerViolation.java
new file mode 100644 (file)
index 0000000..0e5772f
--- /dev/null
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/DataListenerViolation.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2017 Pantheon Technologies, s.r.o. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.clustering.it.provider.impl;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Optional;
+import org.opendaylight.controller.clustering.it.provider.NormalizedNodeDiff;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+public final class DataListenerViolation {
+    private final NormalizedNode<?, ?> expected;
+    private final NormalizedNode<?, ?> actual;
+    private final long sequence;
+
+    DataListenerViolation(final long sequence, final NormalizedNode<?, ?> expected, final NormalizedNode<?, ?> actual) {
+        this.sequence = sequence;
+        this.expected = expected;
+        this.actual = actual;
+    }
+
+    public long getSequence() {
+        return sequence;
+    }
+
+    public Optional<NormalizedNodeDiff> toDiff() {
+        return NormalizedNodeDiff.compute(expected, actual);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this).add("sequence", sequence).toString();
+    }
+
+}

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsDOMDataTreeLIstener.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsDOMDataTreeLIstener.java
index f5b55fd0775eaf3f3442fcc98de9c06d8b467ba1..080a23a3349942d735f35966d396dd8b6bf070e5 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsDOMDataTreeLIstener.java
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsDOMDataTreeLIstener.java
@@ -8,104 +8,24 @@
 
 package org.opendaylight.controller.clustering.it.provider.impl;
 
-import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.SettableFuture;
 import java.util.Collection;
 import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import javax.annotation.Nonnull;
 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
 import org.opendaylight.mdsal.dom.api.DOMDataTreeListeningException;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-public class IdIntsDOMDataTreeLIstener implements DOMDataTreeListener {
-
-    private static final Logger LOG = LoggerFactory.getLogger(IdIntsDOMDataTreeLIstener.class);
-    private static final long SECOND_AS_NANO = 1000000000;
-
-    private NormalizedNode<?, ?> localCopy = null;
-    private AtomicLong lastNotifTimestamp = new AtomicLong(0);
-    private ScheduledFuture<?> scheduledFuture;
-    private ScheduledExecutorService executorService;
-
-    @Override
+public class IdIntsDOMDataTreeLIstener extends AbstractDataListener implements DOMDataTreeListener {
+     @Override
     public void onDataTreeChanged(@Nonnull final Collection<DataTreeCandidate> changes,
                                   @Nonnull final Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>> subtrees) {
-
-        // There should only be one candidate reported
-        Preconditions.checkState(changes.size() == 1);
-
-        lastNotifTimestamp.set(System.nanoTime());
-
-        // do not log the change into debug, only use trace since it will lead to OOM on default heap settings
-        LOG.debug("Received data tree changed");
-
-        changes.forEach(change -> {
-            if (change.getRootNode().getDataAfter().isPresent()) {
-                LOG.trace("Received change, data before: {}, data after: {}",
-                        change.getRootNode().getDataBefore().isPresent()
-                                ? change.getRootNode().getDataBefore().get() : "",
-                        change.getRootNode().getDataAfter().get());
-
-                if (localCopy == null || checkEqual(change.getRootNode().getDataBefore().get())) {
-                    localCopy = change.getRootNode().getDataAfter().get();
-                } else {
-                    LOG.warn("Ignoring notification.");
-                    LOG.trace("Ignored notification content: {}", change);
-                }
-            } else {
-                LOG.warn("getDataAfter() is missing from notification. change: {}", change);
-            }
-        });
+        onReceivedChanges(changes);
     }
 
     @Override
-    public void onDataTreeFailed(@Nonnull Collection<DOMDataTreeListeningException> causes) {
-
-    }
-
-    public boolean hasTriggered() {
-        return localCopy != null;
+    public void onDataTreeFailed(@Nonnull final Collection<DOMDataTreeListeningException> causes) {
+        onReceivedError(causes);
     }
-
-    public Future<Void> tryFinishProcessing() {
-        executorService = Executors.newSingleThreadScheduledExecutor();
-        final SettableFuture<Void> settableFuture = SettableFuture.create();
-
-        scheduledFuture = executorService.scheduleAtFixedRate(new CheckFinishedTask(settableFuture), 0, 1, TimeUnit.SECONDS);
-        return settableFuture;
-    }
-
-    public boolean checkEqual(final NormalizedNode<?, ?> expected) {
-        return localCopy.equals(expected);
-    }
-
-    private class CheckFinishedTask implements Runnable {
-
-        private final SettableFuture<Void> future;
-
-        CheckFinishedTask(final SettableFuture<Void> future) {
-            this.future = future;
-        }
-
-        @Override
-        public void run() {
-            if (System.nanoTime() - lastNotifTimestamp.get() > (SECOND_AS_NANO * 4)) {
-                scheduledFuture.cancel(false);
-                future.set(null);
-
-                executorService.shutdown();
-            }
-        }
-    }
-
 }

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsListener.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsListener.java
index f98822b6f70f020e65d4235337770be49c270491..0ab7830760b8f980cbd90df740f6e5aba9e0ba7d 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsListener.java
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/IdIntsListener.java
@@ -8,94 +8,14 @@
 
 package org.opendaylight.controller.clustering.it.provider.impl;
 
-import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.SettableFuture;
 import java.util.Collection;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import javax.annotation.Nonnull;
 import org.opendaylight.controller.md.sal.dom.api.ClusteredDOMDataTreeChangeListener;
-import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class IdIntsListener implements ClusteredDOMDataTreeChangeListener {
-
-    private static final Logger LOG = LoggerFactory.getLogger(IdIntsListener.class);
-    private static final long SECOND_AS_NANO = 1000000000;
-
-    private NormalizedNode<?, ?> localCopy = null;
-    private AtomicLong lastNotifTimestamp = new AtomicLong(0);
-    private ScheduledExecutorService executorService;
-    private ScheduledFuture<?> scheduledFuture;
 
+public class IdIntsListener extends AbstractDataListener implements ClusteredDOMDataTreeChangeListener {
     @Override
     public void onDataTreeChanged(@Nonnull final Collection<DataTreeCandidate> changes) {
-
-        // There should only be one candidate reported
-        Preconditions.checkState(changes.size() == 1);
-
-        lastNotifTimestamp.set(System.nanoTime());
-
-        // do not log the change into debug, only use trace since it will lead to OOM on default heap settings
-        LOG.debug("Received data tree changed");
-
-        changes.forEach(change -> {
-            if (change.getRootNode().getDataAfter().isPresent()) {
-                LOG.trace("Received change, data before: {}, data after: ",
-                        change.getRootNode().getDataBefore().isPresent()
-                                ? change.getRootNode().getDataBefore().get() : "",
-                        change.getRootNode().getDataAfter().get());
-
-                if (localCopy == null || checkEqual(change.getRootNode().getDataBefore().get())) {
-                    localCopy = change.getRootNode().getDataAfter().get();
-                } else {
-                    LOG.warn("Ignoring notification.");
-                    LOG.trace("Ignored notification content: {}", change);
-                }
-            } else {
-                LOG.warn("getDataAfter() is missing from notification. change: {}", change);
-            }
-        });
-    }
-
-    public boolean hasTriggered() {
-        return localCopy != null;
-    }
-
-    public boolean checkEqual(final NormalizedNode<?, ?> expected) {
-        return localCopy.equals(expected);
-    }
-
-    public Future<Void> tryFinishProcessing() {
-        executorService = Executors.newSingleThreadScheduledExecutor();
-        final SettableFuture<Void> settableFuture = SettableFuture.create();
-
-        scheduledFuture = executorService.scheduleAtFixedRate(new CheckFinishedTask(settableFuture), 0, 1, TimeUnit.SECONDS);
-        return settableFuture;
-    }
-
-    private class CheckFinishedTask implements Runnable {
-
-        private final SettableFuture<Void> future;
-
-        public CheckFinishedTask(final SettableFuture<Void> future) {
-            this.future = future;
-        }
-
-        @Override
-        public void run() {
-            if (System.nanoTime() - lastNotifTimestamp.get() > SECOND_AS_NANO * 4) {
-                scheduledFuture.cancel(false);
-                future.set(null);
-
-                executorService.shutdown();
-            }
-        }
+        onReceivedChanges(changes);
     }
 }