Bug 7800 - Implement agent RPCs for Yang notification testing 20/52320/10
authorTomas Cere <tcere@cisco.com>
Mon, 27 Feb 2017 15:32:56 +0000 (16:32 +0100)
committerTom Pantelis <tpanteli@brocade.com>
Mon, 13 Mar 2017 13:16:20 +0000 (13:16 +0000)
Change-Id: Ia09ada21a737714a0feaa6be4a019ee1703b6d52
Signed-off-by: Tomas Cere <tcere@cisco.com>
opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/MdsalLowLevelTestProvider.java
opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/PublishNotificationsTask.java [new file with mode: 0644]
opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/YnlListener.java [new file with mode: 0644]
opendaylight/md-sal/samples/clustering-test-app/provider/src/main/resources/org/opendaylight/blueprint/cluster-test-app.xml

index bdc2d2f..7ada4be 100644 (file)
@@ -9,22 +9,24 @@
 package org.opendaylight.controller.clustering.it.provider;
 
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.Future;
 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.PublishNotificationsTask;
 import org.opendaylight.controller.clustering.it.provider.impl.RoutedGetConstantService;
 import org.opendaylight.controller.clustering.it.provider.impl.SingletonGetConstantService;
-import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodec;
-import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodecFactory;
+import org.opendaylight.controller.clustering.it.provider.impl.YnlListener;
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
+import org.opendaylight.controller.md.sal.binding.api.NotificationService;
 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationRegistration;
 import org.opendaylight.controller.md.sal.dom.api.DOMRpcProviderService;
 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
 import org.opendaylight.controller.sal.core.api.model.SchemaService;
 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
-import org.opendaylight.mdsal.binding.generator.impl.GeneratedClassLoadingStrategy;
 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
 import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.AddShardReplicaInput;
@@ -54,7 +56,6 @@ import org.opendaylight.yangtools.yang.common.RpcError;
 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
 import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
-import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,6 +66,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     private final RpcProviderRegistry rpcRegistry;
     private final BindingAwareBroker.RpcRegistration<OdlMdsalLowlevelControlService> registration;
     private final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer;
+    private final NotificationPublishService notificationPublishService;
+    private final NotificationService notificationService;
     private final SchemaService schemaService;
     private final ClusterSingletonServiceProvider singletonService;
     private final DOMRpcProviderService domRpcService;
@@ -72,6 +75,8 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     private Map<InstanceIdentifier<?>, DOMRpcImplementationRegistration<RoutedGetConstantService>> routedRegistrations =
             new HashMap<>();
 
+    private Map<String, ListenerRegistration<YnlListener>> ynlRegistrations = new HashMap<>();
+
     private DOMRpcImplementationRegistration<GetConstantService> globalGetConstantRegistration = null;
     private ClusterSingletonServiceRegistration getSingletonConstantRegistration;
     private FlappingSingletonService flappingSingletonService;
@@ -80,12 +85,16 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
                                      final DOMRpcProviderService domRpcService,
                                      final ClusterSingletonServiceProvider singletonService,
                                      final SchemaService schemaService,
-                                     final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer) {
+                                     final BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer,
+                                     final NotificationPublishService notificationPublishService,
+                                     final NotificationService notificationService) {
         this.rpcRegistry = rpcRegistry;
         this.domRpcService = domRpcService;
         this.singletonService = singletonService;
         this.schemaService = schemaService;
         this.bindingNormalizedNodeSerializer = bindingNormalizedNodeSerializer;
+        this.notificationPublishService = notificationPublishService;
+        this.notificationService = notificationService;
 
         registration = rpcRegistry.addRpcImplementation(OdlMdsalLowlevelControlService.class, this);
     }
@@ -117,8 +126,16 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     }
 
     @Override
-    public Future<RpcResult<Void>> publishNotifications(PublishNotificationsInput input) {
-        return null;
+    public Future<RpcResult<Void>> publishNotifications(final PublishNotificationsInput input) {
+        LOG.debug("publish-notifications, input: {}", input);
+
+        final PublishNotificationsTask task = new PublishNotificationsTask(notificationPublishService, input.getId(),
+                input.getSeconds(), input.getNotificationsPerSecond());
+
+        final SettableFuture<RpcResult<Void>> settableFuture = SettableFuture.create();
+        task.start(settableFuture);
+
+        return settableFuture;
     }
 
     @Override
@@ -147,8 +164,20 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     }
 
     @Override
-    public Future<RpcResult<Void>> subscribeYnl(SubscribeYnlInput input) {
-        return null;
+    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());
+        }
+
+        ynlRegistrations.put(input.getId(),
+                notificationService.registerNotificationListener(new YnlListener(input.getId())));
+
+        return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
     }
 
     @Override
@@ -300,8 +329,23 @@ public class MdsalLowLevelTestProvider implements OdlMdsalLowlevelControlService
     }
 
     @Override
-    public Future<RpcResult<UnsubscribeYnlOutput>> unsubscribeYnl(UnsubscribeYnlInput input) {
-        return null;
+    public Future<RpcResult<UnsubscribeYnlOutput>> unsubscribeYnl(final UnsubscribeYnlInput input) {
+        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);
+        }
+
+        final ListenerRegistration<YnlListener> registration = ynlRegistrations.remove(input.getId());
+        final UnsubscribeYnlOutput output = registration.getInstance().getOutput();
+
+        registration.close();
+
+        return Futures.immediateFuture(RpcResultBuilder.<UnsubscribeYnlOutput>success().withResult(output).build());
     }
 
     @Override
diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/PublishNotificationsTask.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/PublishNotificationsTask.java
new file mode 100644 (file)
index 0000000..685831d
--- /dev/null
@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2017 Cisco Systems, Inc. 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.Preconditions;
+import com.google.common.util.concurrent.SettableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequence;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequenceBuilder;
+import org.opendaylight.yangtools.yang.common.RpcError;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PublishNotificationsTask implements Runnable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PublishNotificationsTask.class);
+    private static final int SECOND_AS_NANO = 1000000000;
+
+    private final NotificationPublishService notificationPublishService;
+    private final String notificationId;
+    private final long timeToTake;
+    private final long delay;
+
+    private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+
+    private long sequenceNumber = 1;
+    private long startTime;
+    private SettableFuture<RpcResult<Void>> completionFuture;
+    private ScheduledFuture<?> scheduledFuture;
+
+    public PublishNotificationsTask(final NotificationPublishService notificationPublishService,
+                                    final String notificationId, final long secondsToTake, final long maxPerSecond) {
+        Preconditions.checkNotNull(notificationPublishService);
+        Preconditions.checkNotNull(notificationId);
+        Preconditions.checkArgument(secondsToTake > 0);
+        Preconditions.checkArgument(maxPerSecond > 0);
+
+        this.notificationPublishService = notificationPublishService;
+        this.notificationId = notificationId;
+        this.timeToTake = secondsToTake * SECOND_AS_NANO;
+        this.delay = SECOND_AS_NANO / maxPerSecond;
+
+        LOG.debug("Delay : {}", delay);
+    }
+
+    @Override
+    public void run() {
+        final long current = System.nanoTime();
+
+        final IdSequence notification =
+                new IdSequenceBuilder().setId(notificationId).setSequenceNumber(sequenceNumber).build();
+        sequenceNumber++;
+
+        try {
+            LOG.debug("Publishing notification: {}", notification);
+            notificationPublishService.putNotification(notification);
+        } catch (final InterruptedException e) {
+            LOG.warn("Unexpected exception while publishing notification, : {}", notification, e);
+            completionFuture.set(RpcResultBuilder.<Void>failed()
+                    .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
+        }
+
+        LOG.debug("current {}, starttime: {}, timetotake: {}, current-start = {}",
+                current, startTime, timeToTake, current - startTime);
+
+        if ((current - startTime) > timeToTake) {
+            completionFuture.set(RpcResultBuilder.<Void>success().build());
+            LOG.debug("Sequence number: {}", sequenceNumber);
+            scheduledFuture.cancel(false);
+            executor.shutdown();
+        }
+    }
+
+    public void start(final SettableFuture<RpcResult<Void>> settableFuture) {
+        startTime = System.nanoTime();
+        completionFuture = settableFuture;
+        scheduledFuture = executor.scheduleAtFixedRate(this, 0, delay, TimeUnit.NANOSECONDS);
+    }
+}
diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/YnlListener.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/YnlListener.java
new file mode 100644 (file)
index 0000000..38e0596
--- /dev/null
@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2017 Cisco Systems, Inc. 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.Preconditions;
+import java.util.concurrent.atomic.AtomicLong;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeYnlOutput;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.control.rev170215.UnsubscribeYnlOutputBuilder;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.IdSequence;
+import org.opendaylight.yang.gen.v1.tag.opendaylight.org._2017.controller.yang.lowlevel.target.rev170215.OdlMdsalLowlevelTargetListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class YnlListener implements OdlMdsalLowlevelTargetListener {
+
+    private static final Logger LOG = LoggerFactory.getLogger(YnlListener.class);
+
+    private final String id;
+
+    private AtomicLong localNumber = new AtomicLong();
+    private AtomicLong allNot = new AtomicLong();
+    private AtomicLong idNot = new AtomicLong();
+    private AtomicLong errNot = new AtomicLong();
+
+    public YnlListener(final String id) {
+        Preconditions.checkNotNull(id);
+        this.id = id;
+    }
+
+    @Override
+    public void onIdSequence(final IdSequence notification) {
+        LOG.debug("Received id-sequence notification, : {}", notification);
+
+        allNot.incrementAndGet();
+
+        if (notification.getId().equals(id)) {
+            idNot.incrementAndGet();
+
+            localNumber.getAndUpdate((value -> {
+                if (notification.getSequenceNumber() - value == 1) {
+                    return value + 1;
+                } else {
+                    errNot.getAndIncrement();
+                    return value;
+                }
+            }));
+        }
+    }
+
+    public UnsubscribeYnlOutput getOutput() {
+        return new UnsubscribeYnlOutputBuilder()
+                .setAllNot(allNot.get())
+                .setErrNot(errNot.get())
+                .setIdNot(idNot.get())
+                .setLocalNumber(localNumber.get())
+                .build();
+    }
+}
index 08a302e..cb7a29a 100644 (file)
@@ -13,6 +13,8 @@
   <reference id="domDataBroker" interface="org.opendaylight.controller.md.sal.dom.api.DOMDataBroker"/>
   <reference id="schemaService" interface="org.opendaylight.controller.sal.core.api.model.SchemaService"/>
   <reference id="normalizedNodeSerializer" interface="org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer"/>
+  <reference id="notificationPublishService" interface="org.opendaylight.controller.md.sal.binding.api.NotificationPublishService" />
+  <reference id="notificationListenerService" interface="org.opendaylight.controller.md.sal.binding.api.NotificationService" />
 
   <bean id="purchaseCarProvider" class="org.opendaylight.controller.clustering.it.provider.PurchaseCarProvider" >
     <property name="notificationProvider" ref="notificationService"/>
@@ -52,6 +54,8 @@
     <argument ref="clusterSingletonService"/>
     <argument ref="schemaService"/>
     <argument ref="normalizedNodeSerializer"/>
+    <argument ref="notificationPublishService"/>
+    <argument ref="notificationListenerService"/>
   </bean>
 
 </blueprint>

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.