gerrit.onap Code Review - ccsdk/features.git/blob

   1 /*
   2  * ============LICENSE_START========================================================================
   3  * ONAP : ccsdk feature sdnr wt
   4  * =================================================================================================
   5  * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
   6  * =================================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   8  * in compliance with the License. You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software distributed under the License
  13  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  14  * or implied. See the License for the specific language governing permissions and limitations under
  15  * the License.
  16  * ============LICENSE_END==========================================================================
  17  */
  18 package org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl;
  19
  20 import java.util.Collection;
  21 import java.util.HashMap;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Optional;
  25 import java.util.concurrent.CopyOnWriteArrayList;
  26 import java.util.concurrent.ExecutorService;
  27 import java.util.concurrent.Executors;
  28 import javax.annotation.Nullable;
  29
  30 import org.eclipse.jdt.annotation.NonNull;
  31 import org.onap.ccsdk.features.sdnr.wt.common.configuration.ConfigurationFileRepresentation;
  32 import org.onap.ccsdk.features.sdnr.wt.common.configuration.filechange.IConfigChangedListener;
  33 import org.onap.ccsdk.features.sdnr.wt.dataprovider.model.IEntityDataProvider;
  34 import org.onap.ccsdk.features.sdnr.wt.dataprovider.model.StatusChangedHandler.StatusKey;
  35 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.NetconfAccessor;
  36 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.NetconfNodeConnectListener;
  37 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.NetconfNodeStateListener;
  38 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.NetconfNodeStateService;
  39 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.VesNotificationListener;
  40 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.NetconfStateConfig;
  41 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.odlAkka.AkkaConfig;
  42 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.odlAkka.ClusterConfig;
  43 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.odlGeo.GeoConfig;
  44 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.rpc.NetconfnodeStateServiceRpcApiImpl;
  45 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.rpc.RpcApigetStateCallback;
  46 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
  47 import org.opendaylight.mdsal.binding.api.DataBroker;
  48 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  49 import org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType;
  50 import org.opendaylight.mdsal.binding.api.DataTreeChangeListener;
  51 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  52 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  53 import org.opendaylight.mdsal.binding.api.MountPoint;
  54 import org.opendaylight.mdsal.binding.api.MountPointService;
  55 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
  56 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  57 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  58 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  59 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.ClusteredConnectionStatus;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.network.topology.topology.topology.types.TopologyNetconf;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.netconfnode.state.rev191011.GetStatusInput;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.netconfnode.state.rev191011.GetStatusOutputBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  67 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  68 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  69 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  70 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  71 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  72 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  73 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  74 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  75 import org.slf4j.Logger;
  76 import org.slf4j.LoggerFactory;
  77
  78 public class NetconfNodeStateServiceImpl
  79         implements NetconfNodeStateService, RpcApigetStateCallback, AutoCloseable, IConfigChangedListener {
  80
  81     private static final Logger LOG = LoggerFactory.getLogger(NetconfNodeStateServiceImpl.class);
  82     private static final String APPLICATION_NAME = "NetconfNodeStateService";
  83     @SuppressWarnings("unused")
  84     private static final String CONFIGURATIONFILE = "etc/netconfnode-status-service.properties";
  85
  86     @SuppressWarnings("null")
  87     private static final @NonNull InstanceIdentifier<Topology> NETCONF_TOPO_IID =
  88             InstanceIdentifier.create(NetworkTopology.class).child(Topology.class,
  89                     new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())));
  90
  91     @SuppressWarnings("null")
  92     private static final @NonNull InstanceIdentifier<Node> NETCONF_NODE_TOPO_IID =
  93             InstanceIdentifier.create(NetworkTopology.class)
  94                     .child(Topology.class, new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())))
  95                     .child(Node.class);
  96
  97     private static final DataTreeIdentifier<Node> NETCONF_NODE_TOPO_TREE_ID =
  98             DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL, NETCONF_NODE_TOPO_IID);
  99
 100     // Name of ODL controller NETCONF instance
 101     private static final NodeId CONTROLLER = new NodeId("controller-config");
 102
 103     // -- OSGi services, provided
 104     private DataBroker dataBroker;
 105     private MountPointService mountPointService;
 106     private RpcProviderService rpcProviderRegistry;
 107     private IEntityDataProvider iEntityDataProvider;
 108     @SuppressWarnings("unused")
 109     private NotificationPublishService notificationPublishService;
 110     @SuppressWarnings("unused")
 111     private ClusterSingletonServiceProvider clusterSingletonServiceProvider;
 112
 113     // -- Parameter
 114     private ListenerRegistration<L1> listenerL1;
 115     private ListenerRegistration<L2> listenerL2;
 116     @SuppressWarnings("unused")
 117     private ClusterSingletonServiceRegistration cssRegistration;
 118
 119     private NetconfnodeStateServiceRpcApiImpl rpcApiService;
 120
 121     /** Indication if init() function called and fully executed **/
 122     private Boolean initializationSuccessful;
 123
 124     /** Manager accessor objects for connection **/
 125     private final NetconfAccessorManager accessorManager;
 126
 127     /** List of all registered listeners **/
 128     private final List<NetconfNodeConnectListener> netconfNodeConnectListenerList;
 129
 130     /** List of all registered listeners **/
 131     private final List<NetconfNodeStateListener> netconfNodeStateListenerList;
 132
 133     /** List of all registered listeners **/
 134     private final List<VesNotificationListener> vesNotificationListenerList;
 135
 136     /** Indicates if running in cluster configuration **/
 137     private boolean isCluster;
 138
 139     /** Indicates the name of the cluster **/
 140     private String clusterName;
 141
 142     /** nodeId to threadPool (size=1) for datatreechange handling) **/
 143     private final Map<String, ExecutorService> handlingPool;
 144
 145     private boolean handleDataTreeAsync;
 146
 147     private ConfigurationFileRepresentation configFileRepresentation;
 148     private NetconfStateConfig config;
 149
 150     /** Blueprint **/
 151     public NetconfNodeStateServiceImpl() {
 152         LOG.info("Creating provider for {}", APPLICATION_NAME);
 153
 154         this.dataBroker = null;
 155         this.mountPointService = null;
 156         this.rpcProviderRegistry = null;
 157         this.notificationPublishService = null;
 158         this.clusterSingletonServiceProvider = null;
 159
 160         this.listenerL1 = null;
 161         this.listenerL2 = null;
 162         this.initializationSuccessful = false;
 163         this.netconfNodeConnectListenerList = new CopyOnWriteArrayList<>();
 164         this.netconfNodeStateListenerList = new CopyOnWriteArrayList<>();
 165         this.vesNotificationListenerList = new CopyOnWriteArrayList<>();
 166         this.accessorManager = new NetconfAccessorManager();
 167         this.handlingPool = new HashMap<>();
 168
 169     }
 170
 171     public void setDataBroker(DataBroker dataBroker) {
 172         this.dataBroker = dataBroker;
 173     }
 174
 175     public void setRpcProviderRegistry(RpcProviderService rpcProviderRegistry) {
 176         this.rpcProviderRegistry = rpcProviderRegistry;
 177     }
 178
 179     public void setNotificationPublishService(NotificationPublishService notificationPublishService) {
 180         this.notificationPublishService = notificationPublishService;
 181     }
 182
 183     public void setMountPointService(MountPointService mountPointService) {
 184         this.mountPointService = mountPointService;
 185     }
 186
 187     public void setClusterSingletonService(ClusterSingletonServiceProvider clusterSingletonService) {
 188         this.clusterSingletonServiceProvider = clusterSingletonService;
 189     }
 190
 191     public void setEntityDataProvider(IEntityDataProvider iEntityDataProvider) {
 192         this.iEntityDataProvider = iEntityDataProvider;
 193     }
 194
 195     /** Blueprint initialization **/
 196     public void init() {
 197
 198         LOG.info("Session Initiated start {}", APPLICATION_NAME);
 199
 200         // Start RPC Service
 201         this.rpcApiService = new NetconfnodeStateServiceRpcApiImpl(rpcProviderRegistry, vesNotificationListenerList);
 202         // Get configuration
 203         this.configFileRepresentation = new ConfigurationFileRepresentation(CONFIGURATIONFILE);
 204         this.config = new NetconfStateConfig(this.configFileRepresentation);
 205         this.handleDataTreeAsync = this.config.handleAsync();
 206         this.configFileRepresentation.registerConfigChangedListener(this);
 207
 208         // Akka setup
 209         AkkaConfig akkaConfig = getAkkaConfig();
 210         this.isCluster = akkaConfig == null ? false : akkaConfig.isCluster();
 211         this.clusterName = akkaConfig == null ? "" : akkaConfig.getClusterConfig().getClusterSeedNodeName("abc");
 212
 213         // Provide status information
 214         ClusterConfig cc = akkaConfig == null ? null : akkaConfig.getClusterConfig();
 215         this.iEntityDataProvider.setStatus(StatusKey.CLUSTER_SIZE,
 216                 cc == null ? "1" : String.format("%d", cc.getClusterSize()));
 217
 218         // RPC Service for specific services
 219         this.rpcApiService.setStatusCallback(this);
 220
 221         LOG.debug("start NetconfSubscriptionManager Service");
 222         //this.netconfChangeListener = new NetconfChangeListener(this, dataBroker);
 223         //this.netconfChangeListener.register();
 224         //DataTreeIdentifier<Node> treeId = new DataTreeIdentifier<>(LogicalDatastoreType.OPERATIONAL, NETCONF_NODE_TOPO_IID);
 225
 226         listenerL1 = dataBroker.registerDataTreeChangeListener(NETCONF_NODE_TOPO_TREE_ID, new L1());
 227         listenerL2 = dataBroker.registerDataTreeChangeListener(NETCONF_NODE_TOPO_TREE_ID, new L2());
 228
 229         this.initializationSuccessful = true;
 230
 231         LOG.info("Session Initiated end. Initialization done {}", initializationSuccessful);
 232
 233     }
 234
 235     /** Blueprint destroy-method method */
 236     public void destroy() {
 237         close();
 238     }
 239
 240     /**
 241      * Getter
 242      *
 243      * @return NetconfnodeStateServiceRpcApiImpl
 244      */
 245     public NetconfnodeStateServiceRpcApiImpl getNetconfnodeStateServiceRpcApiImpl() {
 246         return rpcApiService;
 247     }
 248
 249     @Override
 250     public GetStatusOutputBuilder getStatus(GetStatusInput input) {
 251         return new GetStatusOutputBuilder();
 252     }
 253
 254     @Override
 255     public <L extends NetconfNodeConnectListener> @NonNull ListenerRegistration<L> registerNetconfNodeConnectListener(
 256             final @NonNull L netconfNodeConnectListener) {
 257         LOG.info("Register connect listener {}", netconfNodeConnectListener.getClass().getName());
 258         netconfNodeConnectListenerList.add(netconfNodeConnectListener);
 259
 260         return new ListenerRegistration<L>() {
 261             @Override
 262             public @NonNull L getInstance() {
 263                 return netconfNodeConnectListener;
 264             }
 265
 266             @Override
 267             public void close() {
 268                 LOG.info("Remove connect listener {}", netconfNodeConnectListener);
 269                 netconfNodeConnectListenerList.remove(netconfNodeConnectListener);
 270             }
 271         };
 272     }
 273
 274     @Override
 275     public <L extends NetconfNodeStateListener> @NonNull ListenerRegistration<L> registerNetconfNodeStateListener(
 276             @NonNull L netconfNodeStateListener) {
 277         LOG.info("Register state listener {}", netconfNodeStateListener.getClass().getName());
 278         netconfNodeStateListenerList.add(netconfNodeStateListener);
 279
 280         return new ListenerRegistration<L>() {
 281             @Override
 282             public @NonNull L getInstance() {
 283                 return netconfNodeStateListener;
 284             }
 285
 286             @Override
 287             public void close() {
 288                 LOG.info("Remove state listener {}", netconfNodeStateListener);
 289                 netconfNodeStateListenerList.remove(netconfNodeStateListener);
 290             }
 291         };
 292     }
 293
 294     @Override
 295     public <L extends VesNotificationListener> @NonNull ListenerRegistration<L> registerVesNotifications(
 296             @NonNull L vesNotificationListener) {
 297         LOG.info("Register Ves notification listener {}", vesNotificationListener.getClass().getName());
 298         vesNotificationListenerList.add(vesNotificationListener);
 299
 300         return new ListenerRegistration<L>() {
 301             @Override
 302             public @NonNull L getInstance() {
 303                 return vesNotificationListener;
 304             }
 305
 306             @Override
 307             public void close() {
 308                 LOG.info("Remove Ves notification listener {}", vesNotificationListener);
 309                 vesNotificationListenerList.remove(vesNotificationListener);
 310             }
 311         };
 312     }
 313
 314     @Override
 315     public void close() {
 316         LOG.info("Closing start ...");
 317         try {
 318             close(rpcApiService, listenerL1, listenerL2);
 319         } catch (Exception e) {
 320             LOG.debug("Closing", e);
 321         }
 322         LOG.info("Closing done");
 323     }
 324
 325     /**
 326      * Used to close all Services, that should support AutoCloseable Pattern
 327      *
 328      * @param toClose
 329      * @throws Exception
 330      */
 331     private void close(AutoCloseable... toCloseList) throws Exception {
 332         for (AutoCloseable element : toCloseList) {
 333             if (element != null) {
 334                 element.close();
 335             }
 336         }
 337         this.configFileRepresentation.unregisterConfigChangedListener(this);
 338     }
 339
 340     /**
 341      * Indication if init() of this bundle successfully done.
 342      *
 343      * @return true if init() was successful. False if not done or not successful.
 344      */
 345     public boolean isInitializationSuccessful() {
 346         return this.initializationSuccessful;
 347     }
 348
 349     /*-------------------------------------------------------------------------------------------
 350      * Functions for interface DeviceManagerService
 351      */
 352
 353     /**
 354      * For each mounted device a mountpoint is created and this listener is called. Mountpoint was created or existing.
 355      * Managed device is now fully connected to node/mountpoint.
 356      *
 357      * @param nNodeId id of the mountpoint
 358      * @param netconfNode mountpoint contents
 359      */
 360     private void enterConnectedState(NodeId nNodeId, NetconfNode netconfNode) {
 361
 362         String mountPointNodeName = nNodeId.getValue();
 363         LOG.info("Access connected state for mountpoint {}", mountPointNodeName);
 364
 365         boolean preConditionMissing = false;
 366         if (mountPointService == null) {
 367             preConditionMissing = true;
 368             LOG.warn("No mountservice available.");
 369         }
 370         if (!initializationSuccessful) {
 371             preConditionMissing = true;
 372             LOG.warn("Devicemanager initialization still pending.");
 373         }
 374         if (preConditionMissing) {
 375             return;
 376         }
 377
 378         boolean isNetconfNodeMaster = isNetconfNodeMaster(netconfNode);
 379         LOG.info("isNetconfNodeMaster indication {} for mountpoint {}", isNetconfNodeMaster, mountPointNodeName);
 380         if (isNetconfNodeMaster) {
 381
 382             InstanceIdentifier<Node> instanceIdentifier =
 383                     NETCONF_TOPO_IID.child(Node.class, new NodeKey(new NodeId(mountPointNodeName)));
 384
 385             Optional<MountPoint> optionalMountPoint = mountPointService.getMountPoint(instanceIdentifier);
 386             if (!optionalMountPoint.isPresent()) {
 387                 LOG.warn("No mountpoint available for Netconf device :: Name : {} ", mountPointNodeName);
 388             } else {
 389                 // Mountpoint is present for sure
 390                 MountPoint mountPoint = optionalMountPoint.get();
 391                 // BindingDOMDataBrokerAdapter.BUILDER_FACTORY;
 392                 LOG.info("Mountpoint with id: {} class:{}", mountPoint.getIdentifier(),
 393                         mountPoint.getClass().getName());
 394
 395                 Optional<DataBroker> optionalNetconfNodeDatabroker = mountPoint.getService(DataBroker.class);
 396
 397                 if (!optionalNetconfNodeDatabroker.isPresent()) {
 398                     LOG.info("Slave mountpoint {} without databroker", mountPointNodeName);
 399                 } else {
 400                     LOG.info("Master mountpoint {}", mountPointNodeName);
 401                     DataBroker netconfNodeDataBroker = optionalNetconfNodeDatabroker.get();
 402                     NetconfAccessor acessor =
 403                             accessorManager.getAccessor(nNodeId, netconfNode, netconfNodeDataBroker, mountPoint);
 404                     /*
 405                      * --> Call Listers for onConnect() Indication
 406                        for (all)
 407                      */
 408                     netconfNodeConnectListenerList.forEach(item -> {
 409                         try {
 410                             item.onEnterConnected(acessor);
 411                         } catch (Exception e) {
 412                             LOG.info("Exception during onEnterConnected listener call", e);
 413                         }
 414                     });
 415
 416                     LOG.info("Connect indication forwarded for {}", mountPointNodeName);
 417                 }
 418             }
 419         }
 420     }
 421
 422     /**
 423      * Leave the connected status to a non connected or removed status for master mountpoint
 424      *
 425      * @param action that occurred
 426      * @param nNodeId id of the mountpoint
 427      * @param netconfNode mountpoint contents or not available on remove
 428      */
 429     private void leaveConnectedState(NodeId nNodeId, Optional<NetconfNode> optionalNetconfNode) {
 430         String mountPointNodeName = nNodeId.getValue();
 431         LOG.info("leaveConnectedState id {}", mountPointNodeName);
 432
 433         if (this.accessorManager.containes(nNodeId)) {
 434             netconfNodeConnectListenerList.forEach(item -> {
 435                 try {
 436                     if (item != null) {
 437                         item.onLeaveConnected(nNodeId, optionalNetconfNode);
 438                     } else {
 439                         LOG.warn("Unexpeced null item during onleave");
 440                     }
 441                 } catch (Exception e) {
 442                     LOG.info("Exception during onLeaveConnected listener call", e);
 443                 }
 444             });
 445             LOG.info("Remove Master mountpoint {}", mountPointNodeName);
 446             this.accessorManager.removeAccessor(nNodeId);
 447         } else {
 448             LOG.info("Master mountpoint already removed {}", mountPointNodeName);
 449         }
 450     }
 451
 452     // ---- onDataTreeChangedHandler
 453
 454     private void handleDataTreeChange(DataObjectModification<Node> root, NodeId nodeId,
 455             ModificationType modificationTyp) {
 456         // Move status into boolean flags for
 457         boolean connectedBefore, connectedAfter, created;
 458         NetconfNode nNodeAfter = getNetconfNode(root.getDataAfter());
 459         connectedAfter = isConnected(nNodeAfter);
 460         if (root.getDataBefore() != null) {
 461             // It is an update or delete
 462             NetconfNode nodeBefore = getNetconfNode(root.getDataBefore());
 463             connectedBefore = isConnected(nodeBefore);
 464             created = false;
 465         } else {
 466             // It is a create
 467             connectedBefore = false;
 468             created = true;
 469         }
 470         LOG.info("L1 NETCONF id:{} t:{} created {} before:{} after:{} akkaIsCluster:{} cl stat:{}", nodeId,
 471                 modificationTyp, created, connectedBefore, connectedAfter, isCluster,
 472                 getClusteredConnectionStatus(nNodeAfter));
 473         switch (modificationTyp) {
 474             case SUBTREE_MODIFIED: // Create or modify sub level node
 475             case WRITE: // Create or modify top level node
 476                 // Treat an overwrite as an update
 477                 // leaveConnected state.before = connected; state.after != connected
 478                 // enterConnected state.after == connected
 479                 // => Here create or update by checking root.getDataBefore() != null
 480                 boolean handled = false;
 481                 if (created) {
 482                     handled = true;
 483                     netconfNodeStateListenerList.forEach(item -> {
 484                         try {
 485                             item.onCreated(nodeId, nNodeAfter);
 486                         } catch (Exception e) {
 487                             LOG.info("Exception during onCreated listener call", e);
 488                         }
 489                     });
 490                 }
 491                 if (!connectedBefore && connectedAfter) {
 492                     handled = true;
 493                     enterConnectedState(nodeId, nNodeAfter);
 494                 }
 495                 if (connectedBefore && !connectedAfter) {
 496                     handled = true;
 497                     leaveConnectedState(nodeId, Optional.of(nNodeAfter));
 498                 }
 499                 if (!handled) {
 500                     //Change if not handled by the messages before
 501                     netconfNodeStateListenerList.forEach(item -> {
 502                         try {
 503                             item.onStateChange(nodeId, nNodeAfter);
 504                         } catch (Exception e) {
 505                             LOG.info("Exception during onStateChange listener call", e);
 506                         }
 507                     });
 508                 }
 509                 // doProcessing(update ? Action.UPDATE : Action.CREATE, nodeId, root);
 510                 break;
 511             case DELETE:
 512                 // Node removed
 513                 // leaveconnected state.before = connected;
 514                 if (!connectedBefore) {
 515                     leaveConnectedState(nodeId, Optional.empty());
 516                 }
 517                 netconfNodeStateListenerList.forEach(item -> {
 518                     try {
 519                         item.onRemoved(nodeId);
 520                     } catch (Exception e) {
 521                         LOG.info("Exception during onRemoved listener call", e);
 522                     }
 523                 });
 524                 // doProcessing(Action.REMOVE, nodeId, root);
 525                 break;
 526         }
 527     }
 528
 529     private void onDataTreeChangedHandler(@NonNull Collection<DataTreeModification<Node>> changes) {
 530         for (final DataTreeModification<Node> change : changes) {
 531
 532             final DataObjectModification<Node> root = change.getRootNode();
 533             if (LOG.isTraceEnabled()) {
 534                 LOG.trace("Handle this modificationType:{} path:{} root:{}", root.getModificationType(),
 535                         change.getRootPath(), root);
 536             }
 537
 538             // Catch potential nullpointer exceptions ..
 539             try {
 540                 ModificationType modificationTyp = root.getModificationType();
 541                 Node node = modificationTyp == ModificationType.DELETE ? root.getDataBefore() : root.getDataAfter();
 542                 NodeId nodeId = node != null ? node.getNodeId() : null;
 543                 if (nodeId == null) {
 544                     LOG.warn("L1 without nodeid.");
 545                 } else {
 546                     if (nodeId.equals(CONTROLLER)) {
 547                         // Do not forward any controller related events to devicemanager
 548                         LOG.debug("Stop processing for [{}]", nodeId);
 549                     } else {
 550                         if (modificationTyp == null) {
 551                             LOG.warn("L1 empty modification type");
 552                         } else {
 553                             if (this.handleDataTreeAsync) {
 554                                 ExecutorService executor = this.handlingPool.getOrDefault(nodeId.getValue(), null);
 555                                 if (executor == null) {
 556                                     executor = Executors.newFixedThreadPool(5);
 557                                     this.handlingPool.put(nodeId.getValue(), executor);
 558                                 }
 559                                 executor.execute(new Thread() {
 560                                     @Override
 561                                     public void run() {
 562                                         handleDataTreeChange(root, nodeId, modificationTyp);
 563                                     }
 564                                 });
 565
 566                             } else {
 567                                 handleDataTreeChange(root, nodeId, modificationTyp);
 568                             }
 569                         }
 570                     }
 571                 }
 572             } catch (NullPointerException | IllegalStateException e) {
 573                 LOG.info("Data not available at ", e);
 574             }
 575         } //for
 576         LOG.info("datatreechanged handler completed");
 577     }
 578
 579     // ---- subclasses for listeners
 580
 581     /**
 582      * Clustered listener function to select the right node from DataObjectModification. Called at all nodes.
 583      */
 584     private class L1 implements ClusteredDataTreeChangeListener<Node> {
 585         @Override
 586         public void onDataTreeChanged(@NonNull Collection<DataTreeModification<Node>> changes) {
 587             LOG.info("L1 TreeChange enter changes:{}", changes.size());
 588             //Debug AkkTimeout NetconfNodeStateServiceImpl.this.pool.execute(new Thread( () -> onDataTreeChangedHandler(changes)));
 589             onDataTreeChangedHandler(changes);
 590             LOG.info("L1 TreeChange leave");
 591         }
 592     }
 593
 594     /**
 595      * Data change, called at leader/master
 596      */
 597     private class L2 implements DataTreeChangeListener<Node> {
 598
 599         @Override
 600         public void onDataTreeChanged(@NonNull Collection<DataTreeModification<Node>> changes) {
 601             LOG.info("L2 TreeChange enter changes:{}", changes.size());
 602             // Do nothing
 603             LOG.info("L2 TreeChange leave");
 604         }
 605     }
 606
 607     /* --- private helpers --- */
 608     private static @Nullable NetconfNode getNetconfNode(Node node) {
 609         return node != null ? node.augmentation(NetconfNode.class) : null;
 610     }
 611
 612     private static boolean isConnected(NetconfNode nNode) {
 613         return nNode != null ? ConnectionStatus.Connected.equals(nNode.getConnectionStatus()) : false;
 614     }
 615
 616     private static @Nullable ClusteredConnectionStatus getClusteredConnectionStatus(NetconfNode node) {
 617         return node != null ? node.getClusteredConnectionStatus() : null;
 618     }
 619
 620     /* -- LOG related functions -- */
 621
 622     /** Analyze configuration **/
 623     private static @Nullable AkkaConfig getAkkaConfig() {
 624         AkkaConfig akkaConfig;
 625         try {
 626             akkaConfig = AkkaConfig.load();
 627             LOG.debug("akka.conf loaded: " + akkaConfig.toString());
 628         } catch (Exception e1) {
 629             akkaConfig = null;
 630             LOG.warn("problem loading akka.conf: " + e1.getMessage());
 631         }
 632         if (akkaConfig != null && akkaConfig.isCluster()) {
 633             LOG.info("cluster mode detected");
 634             if (GeoConfig.fileExists()) {
 635                 try {
 636                     LOG.debug("try to load geoconfig");
 637                     GeoConfig.load();
 638                 } catch (Exception err) {
 639                     LOG.warn("problem loading geoconfig: " + err.getMessage());
 640                 }
 641             } else {
 642                 LOG.debug("no geoconfig file found");
 643             }
 644         } else {
 645             LOG.info("single node mode detected");
 646         }
 647         return akkaConfig;
 648     }
 649
 650     private boolean isNetconfNodeMaster(NetconfNode nNode) {
 651         if (this.isCluster) {
 652             LOG.debug("check if me is responsible for node");
 653             ClusteredConnectionStatus ccs = nNode.getClusteredConnectionStatus();
 654             @SuppressWarnings("null")
 655             @NonNull
 656             String masterNodeName =
 657                     ccs == null || ccs.getNetconfMasterNode() == null ? "null" : ccs.getNetconfMasterNode();
 658             LOG.debug("sdnMasterNode=" + masterNodeName + " and sdnMyNode=" + clusterName);
 659             if (!masterNodeName.equals(clusterName)) {
 660                 LOG.debug("netconf change but me is not master for this node");
 661                 return false;
 662             }
 663         }
 664         return true;
 665     }
 666
 667     @Override
 668     public void onConfigChanged() {
 669         this.handleDataTreeAsync = this.config.handleAsync();
 670
 671     }
 672 }