Fix notification subscription issue

Many devices do not advertise netmod capabilities,
which causes error for subscription.

- fix code to remove netmod dependency
- refactor networkmodel NetConfTopologyListener

JIRA: TRNSPRTPCE-621
Signed-off-by: Balagangadhar Bathula <bb4341@att.com>
Change-Id: Ieebf7d893fb6050af3ca89db68ca47f145014211
Co-authored-by: guillaume.lambert <guillaume.lambert@orange.com>

diff --git a/networkmodel/src/main/java/org/opendaylight/transportpce/networkmodel/NetConfTopologyListener.java b/networkmodel/src/main/java/org/opendaylight/transportpce/networkmodel/NetConfTopologyListener.java
index b1b2cc0b0f25e841639337f862cf83241f2ceac6..dedc387a0d7c400eb80054f59246a68c34caa8f7 100644 (file)
--- a/networkmodel/src/main/java/org/opendaylight/transportpce/networkmodel/NetConfTopologyListener.java
+++ b/networkmodel/src/main/java/org/opendaylight/transportpce/networkmodel/NetConfTopologyListener.java
@@ -10,6 +10,7 @@ package org.opendaylight.transportpce.networkmodel;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ListenableFuture;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
@@ -22,9 +23,7 @@ import org.opendaylight.mdsal.binding.api.DataTreeModification;
 import org.opendaylight.mdsal.binding.api.MountPoint;
 import org.opendaylight.mdsal.binding.api.NotificationService;
 import org.opendaylight.mdsal.binding.api.RpcConsumerRegistry;
-import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
 import org.opendaylight.transportpce.common.StringConstants;
-import org.opendaylight.transportpce.common.Timeouts;
 import org.opendaylight.transportpce.common.device.DeviceTransactionManager;
 import org.opendaylight.transportpce.common.mapping.PortMapping;
 import org.opendaylight.transportpce.networkmodel.dto.NodeRegistration;
@@ -33,14 +32,10 @@ import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification.
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionOutput;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.NotificationsService;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.Netconf;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.Streams;
-import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.Stream;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.available.capabilities.AvailableCapability;
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,8 +50,11 @@ public class NetConfTopologyListener implements DataTreeChangeListener<Node> {
     private final Map<String, NodeRegistration> registrations;
     private final PortMapping portMapping;
 
-    public NetConfTopologyListener(final NetworkModelService networkModelService, final DataBroker dataBroker,
-             DeviceTransactionManager deviceTransactionManager, PortMapping portMapping) {
+    public NetConfTopologyListener(
+            final NetworkModelService networkModelService,
+            final DataBroker dataBroker,
+            DeviceTransactionManager deviceTransactionManager,
+            PortMapping portMapping) {
         this.networkModelService = networkModelService;
         this.dataBroker = dataBroker;
         this.deviceTransactionManager = deviceTransactionManager;
@@ -83,24 +81,24 @@ public class NetConfTopologyListener implements DataTreeChangeListener<Node> {
                 case WRITE:
                     NetconfNode netconfNodeAfter = rootNode.getDataAfter().augmentation(NetconfNode.class);
                     if (ConnectionStatus.Connecting.equals(netconfNodeBefore.getConnectionStatus())
-                        && ConnectionStatus.Connected.equals(netconfNodeAfter.getConnectionStatus())) {
+                            && ConnectionStatus.Connected.equals(netconfNodeAfter.getConnectionStatus())) {
                         LOG.info("Connecting Node: {}", nodeId);
-                        Optional<AvailableCapability> deviceCapabilityOpt = netconfNodeAfter
-                            .getAvailableCapabilities().getAvailableCapability().stream()
-                            .filter(cp -> cp.getCapability().contains(StringConstants.OPENROADM_DEVICE_MODEL_NAME))
-                            .sorted((c1, c2) -> c2.getCapability().compareTo(c1.getCapability()))
-                            .findFirst();
+                        Optional<AvailableCapability> deviceCapabilityOpt =
+                            netconfNodeAfter.getAvailableCapabilities().getAvailableCapability().stream()
+                                .filter(cp -> cp.getCapability().contains(StringConstants.OPENROADM_DEVICE_MODEL_NAME))
+                                .sorted((c1, c2) -> c2.getCapability().compareTo(c1.getCapability()))
+                                .findFirst();
                         if (deviceCapabilityOpt.isEmpty()) {
                             LOG.error("Unable to get openroadm-device-capability");
                             return;
                         }
                         this.networkModelService
                             .createOpenRoadmNode(nodeId, deviceCapabilityOpt.get().getCapability());
-                        onDeviceConnected(nodeId,deviceCapabilityOpt.get().getCapability());
+                        onDeviceConnected(nodeId, deviceCapabilityOpt.get().getCapability());
                         LOG.info("Device {} correctly connected to controller", nodeId);
                     }
                     if (ConnectionStatus.Connected.equals(netconfNodeBefore.getConnectionStatus())
-                        && ConnectionStatus.Connecting.equals(netconfNodeAfter.getConnectionStatus())) {
+                            && ConnectionStatus.Connecting.equals(netconfNodeAfter.getConnectionStatus())) {
                         LOG.warn("Node: {} is being disconnected", nodeId);
                     }
                     break;
@@ -114,36 +112,33 @@ public class NetConfTopologyListener implements DataTreeChangeListener<Node> {
     private void onDeviceConnected(final String nodeId, String openRoadmVersion) {
         LOG.info("onDeviceConnected: {}", nodeId);
         Optional<MountPoint> mountPointOpt = this.deviceTransactionManager.getDeviceMountPoint(nodeId);
-        MountPoint mountPoint;
-        if (mountPointOpt.isPresent()) {
-            mountPoint = mountPointOpt.get();
-        } else {
+        if (mountPointOpt.isEmpty()) {
             LOG.error("Failed to get mount point for node {}", nodeId);
             return;
         }
+        MountPoint mountPoint = mountPointOpt.get();
         final Optional<NotificationService> notificationService = mountPoint.getService(NotificationService.class);
-        if (!notificationService.isPresent()) {
+        if (notificationService.isEmpty()) {
             LOG.error(RPC_SERVICE_FAILED, nodeId);
             return;
         }
-        NodeRegistration nodeRegistration = new NodeRegistration(nodeId, openRoadmVersion,
-            notificationService.get(), this.dataBroker, this.portMapping);
+        NodeRegistration nodeRegistration =
+            new NodeRegistration(
+                nodeId, openRoadmVersion, notificationService.get(), this.dataBroker, this.portMapping);
         nodeRegistration.registerListeners();
         registrations.put(nodeId, nodeRegistration);
-        String streamName = getSupportedStream(nodeId);
-        LOG.info("Device is supporting notification stream {}",streamName);
-        subscribeStream(mountPoint, streamName, nodeId);
+
+        subscribeStream(mountPoint, nodeId);
     }
 
     private void onDeviceDisConnected(final String nodeId) {
         LOG.info("onDeviceDisConnected: {}", nodeId);
-        NodeRegistration nodeRegistration = this.registrations.remove(nodeId);
-        nodeRegistration.unregisterListeners();
+        this.registrations.remove(nodeId).unregisterListeners();
     }
 
-    private boolean subscribeStream(MountPoint mountPoint, String streamName, String nodeId) {
+    private boolean subscribeStream(MountPoint mountPoint, String nodeId) {
         final Optional<RpcConsumerRegistry> service = mountPoint.getService(RpcConsumerRegistry.class);
-        if (!service.isPresent()) {
+        if (service.isEmpty()) {
             return false;
         }
         final NotificationsService rpcService = service.get().getRpcService(NotificationsService.class);
@@ -151,23 +146,28 @@ public class NetConfTopologyListener implements DataTreeChangeListener<Node> {
             LOG.error(RPC_SERVICE_FAILED, nodeId);
             return false;
         }
-        final CreateSubscriptionInputBuilder createSubscriptionInputBuilder = new CreateSubscriptionInputBuilder()
-            .setStream(new StreamNameType(streamName));
-        LOG.info("Triggering notification stream {} for node {}", streamName, nodeId);
-        ListenableFuture<RpcResult<CreateSubscriptionOutput>> subscription = rpcService
-            .createSubscription(createSubscriptionInputBuilder.build());
-        try {
-            subscription.get();
-        } catch (InterruptedException | ExecutionException e) {
-            LOG.error("Error during subscription to stream {}", streamName, e);
+        // Set the default stream as OPENROADM
+        for (String streamName : List.of("OPENROADM", "NETCONF")) {
+            LOG.info("Triggering notification stream {} for node {}", streamName, nodeId);
+            ListenableFuture<RpcResult<CreateSubscriptionOutput>> subscription =
+                rpcService.createSubscription(
+                    new CreateSubscriptionInputBuilder().setStream(new StreamNameType(streamName)).build());
+            // If OpenROADM stream is not supported
+            // Try NETCONF subscription
+            if (checkSupportedStream(streamName, subscription)) {
+                return true;
+            }
         }
-        return true;
+        return false;
     }
 
     @VisibleForTesting
-    public NetConfTopologyListener(final NetworkModelService networkModelService, final DataBroker dataBroker,
-        DeviceTransactionManager deviceTransactionManager, PortMapping portMapping,
-        Map<String, NodeRegistration> registrations) {
+    public NetConfTopologyListener(
+            final NetworkModelService networkModelService,
+            final DataBroker dataBroker,
+            DeviceTransactionManager deviceTransactionManager,
+            PortMapping portMapping,
+            Map<String, NodeRegistration> registrations) {
         this.networkModelService = networkModelService;
         this.dataBroker = dataBroker;
         this.deviceTransactionManager = deviceTransactionManager;
@@ -175,21 +175,18 @@ public class NetConfTopologyListener implements DataTreeChangeListener<Node> {
         this.registrations = registrations;
     }
 
-    private String getSupportedStream(String nodeId) {
-        InstanceIdentifier<Streams> streamsIID = InstanceIdentifier.create(Netconf.class).child(Streams.class);
-        Optional<Streams> ordmInfoObject =
-                deviceTransactionManager.getDataFromDevice(nodeId, LogicalDatastoreType.OPERATIONAL, streamsIID,
-                        Timeouts.DEVICE_READ_TIMEOUT, Timeouts.DEVICE_READ_TIMEOUT_UNIT);
-        if (ordmInfoObject == null || ordmInfoObject.isEmpty() || ordmInfoObject.get().getStream().isEmpty()) {
-            LOG.error("List of streams supports by device is not present");
-            return "NETCONF";
-        }
-        for (Stream strm : ordmInfoObject.get().getStream().values()) {
-            LOG.debug("Streams are {}", strm);
-            if ("OPENROADM".equalsIgnoreCase(strm.getName().getValue())) {
-                return strm.getName().getValue();
-            }
+    private boolean checkSupportedStream(
+            String streamName,
+            ListenableFuture<RpcResult<CreateSubscriptionOutput>> subscription) {
+        boolean subscriptionSuccessful = false;
+        try {
+            // Using if condition does not work, since we need to handle exceptions
+            subscriptionSuccessful = subscription.get().isSuccessful();
+            LOG.info("{} subscription is {}", streamName, subscriptionSuccessful);
+        } catch (InterruptedException | ExecutionException e) {
+            LOG.error("Error during subscription to stream {}", streamName, e);
         }
-        return "NETCONF";
+        return subscriptionSuccessful;
     }
+
 }