Bug 6348 : car:stop-stress-test RPC to return success & failure counters

[controller.git] / opendaylight / md-sal / samples / clustering-test-app / provider / src / main / java / org / opendaylight / controller / clustering / it / provider / CarProvider.java

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java
index 1ab7f5e9016c30a29ab493c978824f6e2b85c44a..69535f7060c7811ef837f8fda9d6c68395e84e53 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/CarProvider.java
@@ -8,19 +8,26 @@
 package org.opendaylight.controller.clustering.it.provider;
 
 import com.google.common.base.Stopwatch;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
+import java.util.Collection;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
+import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
+import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
 import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
+import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.CarId;
@@ -28,10 +35,13 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controll
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.Cars;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.CarsBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.RegisterOwnershipInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.StopStressTestOutput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.StopStressTestOutputBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.StressTestInput;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.UnregisterOwnershipInput;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.cars.CarEntry;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.car.rev140818.cars.CarEntryBuilder;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
 import org.opendaylight.yangtools.yang.common.RpcResult;
@@ -51,12 +61,24 @@ public class CarProvider implements CarService {
 
     private static final String ENTITY_TYPE = "cars";
 
+    private AtomicLong succcessCounter = new AtomicLong();
+    private AtomicLong failureCounter = new AtomicLong();
+
     private final CarEntityOwnershipListener ownershipListener = new CarEntityOwnershipListener();
     private final AtomicBoolean registeredListener = new AtomicBoolean();
 
     private volatile Thread testThread;
     private volatile boolean stopThread;
 
+    private static final InstanceIdentifier CARS_IID = InstanceIdentifier.builder(Cars.class).build();
+    private static final DataTreeIdentifier<Cars> CARS_DTID = new DataTreeIdentifier<>(
+            LogicalDatastoreType.CONFIGURATION, CARS_IID);
+
+    private Collection<ListenerRegistration<DataChangeListener>> carsDclRegistrations =
+            Sets.newConcurrentHashSet();
+    private Collection<ListenerRegistration<CarDataTreeChangeListener>> carsDtclRegistrations =
+            Sets.newConcurrentHashSet();
+
     public CarProvider(DataBroker dataProvider, EntityOwnershipService ownershipService) {
         this.dataProvider = dataProvider;
         this.ownershipService = ownershipService;
@@ -82,8 +104,8 @@ public class CarProvider implements CarService {
         if ((input.getRate() == null) || (input.getRate() == 0)) {
             log.info("Exiting stress test as no rate is given.");
             return Futures.immediateFuture(RpcResultBuilder.<Void>failed()
-                                           .withError(ErrorType.PROTOCOL, "invalid rate")
-                                           .build());
+                    .withError(ErrorType.PROTOCOL, "invalid rate")
+                    .build());
         } else {
             inputRate = input.getRate();
         }
@@ -97,6 +119,9 @@ public class CarProvider implements CarService {
         log.info("Stress test starting : rate: {} count: {}", inputRate, inputCount);
 
         stopThread();
+        // clear counters
+        succcessCounter.set(0);
+        failureCounter.set(0);
 
         WriteTransaction tx = dataProvider.newWriteOnlyTransaction();
         InstanceIdentifier<Cars> carsId = InstanceIdentifier.<Cars>builder(Cars.class).build();
@@ -123,7 +148,22 @@ public class CarProvider implements CarService {
                     tx.put(LogicalDatastoreType.CONFIGURATION,
                             InstanceIdentifier.<Cars>builder(Cars.class).child(CarEntry.class, car.getKey()).build(),
                             car);
-                    tx.submit();
+                    CheckedFuture<Void, TransactionCommitFailedException> future =  tx.submit();
+                    Futures.addCallback(future, new FutureCallback<Void>() {
+
+                        @Override
+                        public void onSuccess(final Void result) {
+                            // Transaction succeeded
+                            succcessCounter.getAndIncrement();
+                        }
+
+                        @Override
+                        public void onFailure(final Throwable t) {
+                            // Transaction failed
+                            failureCounter.getAndIncrement();
+                            LOG.error("Put Cars failed", t);
+                        }
+                    });
                     try {
                         TimeUnit.NANOSECONDS.sleep(sleep);
                     } catch (InterruptedException e) {
@@ -149,9 +189,20 @@ public class CarProvider implements CarService {
     }
 
     @Override
-    public Future<RpcResult<Void>> stopStressTest() {
+    public Future<RpcResult<StopStressTestOutput>> stopStressTest() {
         stopThread();
-        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
+        StopStressTestOutputBuilder stopStressTestOutput;
+        stopStressTestOutput = new StopStressTestOutputBuilder()
+                .setSuccessCount(succcessCounter.longValue())
+                .setFailureCount(failureCounter.longValue());
+
+        StopStressTestOutput result = stopStressTestOutput.build();
+        log.info("Executed Stop Stress test; No. of cars created {}; " +
+                "No. of cars failed {}; ", succcessCounter, failureCounter);
+        // clear counters
+        succcessCounter.set(0);
+        failureCounter.set(0);
+        return Futures.immediateFuture(RpcResultBuilder.<StopStressTestOutput>success(result).build());
     }
 
 
@@ -183,4 +234,61 @@ public class CarProvider implements CarService {
             LOG.info("ownershipChanged: {}", ownershipChange);
         }
     }
+
+    @Override
+    public Future<RpcResult<java.lang.Void>> registerLoggingDcl() {
+        LOG.info("Registering a new CarDataChangeListener");
+        final ListenerRegistration carsDclRegistration = dataProvider.registerDataChangeListener(
+                LogicalDatastoreType.CONFIGURATION, CARS_IID, new CarDataChangeListener(),
+                AsyncDataBroker.DataChangeScope.SUBTREE);
+
+        if (carsDclRegistration != null) {
+            carsDclRegistrations.add(carsDclRegistration);
+            return RpcResultBuilder.<Void>success().buildFuture();
+        }
+        return RpcResultBuilder.<Void>failed().buildFuture();
+    }
+
+    @Override
+    public Future<RpcResult<java.lang.Void>> registerLoggingDtcl() {
+        LOG.info("Registering a new CarDataTreeChangeListener");
+        final ListenerRegistration<CarDataTreeChangeListener> carsDtclRegistration =
+                dataProvider.registerDataTreeChangeListener(CARS_DTID, new CarDataTreeChangeListener());
+
+        if (carsDtclRegistration != null) {
+            carsDtclRegistrations.add(carsDtclRegistration);
+            return RpcResultBuilder.<Void>success().buildFuture();
+        }
+        return RpcResultBuilder.<Void>failed().buildFuture();
+    }
+
+    @Override
+    public Future<RpcResult<java.lang.Void>> unregisterLoggingDcls() {
+        LOG.info("Unregistering the CarDataChangeListener(s)");
+        synchronized (carsDclRegistrations) {
+            int numListeners = 0;
+            for (ListenerRegistration<DataChangeListener> carsDclRegistration : carsDclRegistrations) {
+                carsDclRegistration.close();
+                numListeners++;
+            }
+            carsDclRegistrations.clear();
+            LOG.info("Unregistered {} CarDataChangeListener(s)", numListeners);
+        }
+        return RpcResultBuilder.<Void>success().buildFuture();
+    }
+
+    @Override
+    public Future<RpcResult<java.lang.Void>> unregisterLoggingDtcls() {
+        LOG.info("Unregistering the CarDataTreeChangeListener(s)");
+        synchronized (carsDtclRegistrations) {
+            int numListeners = 0;
+            for (ListenerRegistration<CarDataTreeChangeListener> carsDtclRegistration : carsDtclRegistrations) {
+                carsDtclRegistration.close();
+                numListeners++;
+            }
+            carsDtclRegistrations.clear();
+            LOG.info("Unregistered {} CaraDataTreeChangeListener(s)", numListeners);
+        }
+        return RpcResultBuilder.<Void>success().buildFuture();
+    }
 }