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.Objects;
  25 import java.util.Optional;
  26 import java.util.concurrent.CopyOnWriteArrayList;
  27 import java.util.concurrent.ExecutorService;
  28 import java.util.concurrent.Executors;
  29 import javax.annotation.Nullable;
  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.access.NetconfAccessorManager;
  41 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.access.NetconfCommunicatorManager;
  42 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.access.dom.DomContext;
  43 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.NetconfStateConfig;
  44 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.odlAkka.AkkaConfig;
  45 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.odlAkka.ClusterConfig;
  46 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.conf.odlGeo.GeoConfig;
  47 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.rpc.NetconfnodeStateServiceRpcApiImpl;
  48 import org.onap.ccsdk.features.sdnr.wt.netconfnodestateservice.impl.rpc.RpcApigetStateCallback;
  49 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
  50 import org.opendaylight.mdsal.binding.api.DataBroker;
  51 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  52 import org.opendaylight.mdsal.binding.api.DataObjectModification.ModificationType;
  53 import org.opendaylight.mdsal.binding.api.DataTreeChangeListener;
  54 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  55 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  56 import org.opendaylight.mdsal.binding.api.MountPointService;
  57 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
  58 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  59 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  60 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  61 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  62 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  63 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNodeConnectionStatus.ConnectionStatus;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.ClusteredConnectionStatus;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.network.topology.topology.topology.types.TopologyNetconf;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.netconfnode.state.rev191011.GetStatusInput;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.netconfnode.state.rev191011.GetStatusOutputBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  71 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  72 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  73 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  74 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  75 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  76 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  77 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  78 import org.opendaylight.yangtools.yang.model.parser.api.YangParserException;
  79 import org.opendaylight.yangtools.yang.model.parser.api.YangParserFactory;
  80 import org.slf4j.Logger;
  81 import org.slf4j.LoggerFactory;
  82
  83 public class NetconfNodeStateServiceImpl
  84         implements NetconfNodeStateService, RpcApigetStateCallback, AutoCloseable, IConfigChangedListener {
  85
  86     private static final Logger LOG = LoggerFactory.getLogger(NetconfNodeStateServiceImpl.class);
  87     private static final String APPLICATION_NAME = "NetconfNodeStateService";
  88     private static final String CONFIGURATIONFILE = "etc/netconfnode-status-service.properties";
  89
  90     private static final @NonNull InstanceIdentifier<Topology> NETCONF_TOPO_IID =
  91             InstanceIdentifier.create(NetworkTopology.class).child(Topology.class,
  92                     new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())));
  93
  94     private static final @NonNull InstanceIdentifier<Node> NETCONF_NODE_TOPO_IID =
  95             InstanceIdentifier.create(NetworkTopology.class)
  96                     .child(Topology.class, new TopologyKey(new TopologyId(TopologyNetconf.QNAME.getLocalName())))
  97                     .child(Node.class);
  98
  99     private static final @NonNull DataTreeIdentifier<Node> NETCONF_NODE_TOPO_TREE_ID =
 100             DataTreeIdentifier.create(LogicalDatastoreType.OPERATIONAL, NETCONF_NODE_TOPO_IID);
 101
 102     // Name of ODL controller NETCONF instance
 103     private static final NodeId CONTROLLER = new NodeId("controller-config");
 104
 105     // -- OSGi services, provided
 106     private DataBroker dataBroker;
 107     private MountPointService mountPointService;
 108     private DOMMountPointService domMountPointService;
 109     private RpcProviderService rpcProviderRegistry;
 110     private IEntityDataProvider iEntityDataProvider;
 111     @SuppressWarnings("unused")
 112     private NotificationPublishService notificationPublishService;
 113     @SuppressWarnings("unused")
 114     private ClusterSingletonServiceProvider clusterSingletonServiceProvider;
 115     private YangParserFactory yangParserFactory;
 116     private BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer;
 117
 118     // -- Parameter
 119     private ListenerRegistration<L1> listenerL1;
 120     private ListenerRegistration<L2> listenerL2;
 121     @SuppressWarnings("unused")
 122     private ClusterSingletonServiceRegistration cssRegistration;
 123
 124     private NetconfnodeStateServiceRpcApiImpl rpcApiService;
 125
 126     /** Indication if init() function called and fully executed **/
 127     private Boolean initializationSuccessful;
 128
 129     /** Manager accessor objects for connection **/
 130     private NetconfAccessorManager accessorManager;
 131
 132     /** List of all registered listeners **/
 133     private final List<NetconfNodeConnectListener> netconfNodeConnectListenerList;
 134
 135     /** List of all registered listeners **/
 136     private final List<NetconfNodeStateListener> netconfNodeStateListenerList;
 137
 138     /** List of all registered listeners **/
 139     private final List<VesNotificationListener> vesNotificationListenerList;
 140
 141     /** Indicates if running in cluster configuration **/
 142     private boolean isCluster;
 143
 144     /** Indicates the name of the cluster **/
 145     private String clusterName;
 146
 147     /** nodeId to threadPool (size=1) for datatreechange handling) **/
 148     private final Map<String, ExecutorService> handlingPool;
 149
 150     private boolean handleDataTreeAsync;
 151
 152     private ConfigurationFileRepresentation configFileRepresentation;
 153     private NetconfStateConfig config;
 154     private NetconfCommunicatorManager netconfCommunicatorManager;
 155     private DomContext domContext;
 156
 157     /** Blueprint **/
 158     public NetconfNodeStateServiceImpl() {
 159         LOG.info("Creating provider for {}", APPLICATION_NAME);
 160
 161         this.dataBroker = null;
 162         this.mountPointService = null;
 163         this.domMountPointService = null;
 164         this.rpcProviderRegistry = null;
 165         this.notificationPublishService = null;
 166         this.clusterSingletonServiceProvider = null;
 167         this.yangParserFactory = null;
 168         this.domContext = null;
 169
 170         this.listenerL1 = null;
 171         this.listenerL2 = null;
 172         this.initializationSuccessful = false;
 173         this.netconfNodeConnectListenerList = new CopyOnWriteArrayList<>();
 174         this.netconfNodeStateListenerList = new CopyOnWriteArrayList<>();
 175         this.vesNotificationListenerList = new CopyOnWriteArrayList<>();
 176         this.accessorManager = null;
 177         this.handlingPool = new HashMap<>();
 178
 179     }
 180
 181     public void setDataBroker(DataBroker dataBroker) {
 182         this.dataBroker = dataBroker;
 183     }
 184
 185     public void setRpcProviderRegistry(RpcProviderService rpcProviderRegistry) {
 186         this.rpcProviderRegistry = rpcProviderRegistry;
 187     }
 188
 189     public void setNotificationPublishService(NotificationPublishService notificationPublishService) {
 190         this.notificationPublishService = notificationPublishService;
 191     }
 192
 193     public void setMountPointService(MountPointService mountPointService) {
 194         this.mountPointService = mountPointService;
 195     }
 196
 197     public void setDomMountPointService(DOMMountPointService domMountPointService) {
 198         this.domMountPointService = domMountPointService;
 199     }
 200
 201     public void setClusterSingletonService(ClusterSingletonServiceProvider clusterSingletonService) {
 202         this.clusterSingletonServiceProvider = clusterSingletonService;
 203     }
 204
 205     public void setEntityDataProvider(IEntityDataProvider iEntityDataProvider) {
 206         this.iEntityDataProvider = iEntityDataProvider;
 207     }
 208
 209     public void setYangParserFactory(YangParserFactory yangParserFactory) {
 210         this.yangParserFactory = yangParserFactory;
 211     }
 212
 213     public void setBindingNormalizedNodeSerializer(BindingNormalizedNodeSerializer bindingNormalizedNodeSerializer) {
 214         this.bindingNormalizedNodeSerializer = bindingNormalizedNodeSerializer;
 215     }
 216
 217     /** Blueprint initialization
 218      * @throws YangParserException **/
 219     public void init() {
 220
 221         LOG.info("Session Initiated start {}", APPLICATION_NAME);
 222         this.domContext = new DomContext(this.yangParserFactory, this.bindingNormalizedNodeSerializer);
 223         this.netconfCommunicatorManager = new NetconfCommunicatorManager(mountPointService, domMountPointService, domContext);
 224         this.accessorManager = new NetconfAccessorManager(netconfCommunicatorManager, domContext);
 225         // Start RPC Service
 226         this.rpcApiService = new NetconfnodeStateServiceRpcApiImpl(rpcProviderRegistry, vesNotificationListenerList);
 227         // Get configuration
 228         this.configFileRepresentation = new ConfigurationFileRepresentation(CONFIGURATIONFILE);
 229         this.config = new NetconfStateConfig(this.configFileRepresentation);
 230         this.handleDataTreeAsync = this.config.handleAsync();
 231         this.configFileRepresentation.registerConfigChangedListener(this);
 232
 233         // Akka setup
 234         AkkaConfig akkaConfig = getAkkaConfig();
 235         this.isCluster = akkaConfig == null ? false : akkaConfig.isCluster();
 236         this.clusterName = akkaConfig == null ? "" : akkaConfig.getClusterConfig().getClusterSeedNodeName("abc");
 237
 238         // Provide status information
 239         ClusterConfig cc = akkaConfig == null ? null : akkaConfig.getClusterConfig();
 240         this.iEntityDataProvider.setStatus(StatusKey.CLUSTER_SIZE,
 241                 cc == null ? "1" : String.format("%d", cc.getClusterSize()));
 242
 243         // RPC Service for specific services
 244         this.rpcApiService.setStatusCallback(this);
 245
 246         LOG.debug("start NetconfSubscriptionManager Service");
 247         //this.netconfChangeListener = new NetconfChangeListener(this, dataBroker);
 248         //this.netconfChangeListener.register();
 249         //DataTreeIdentifier<Node> treeId = new DataTreeIdentifier<>(LogicalDatastoreType.OPERATIONAL, NETCONF_NODE_TOPO_IID);
 250
 251         listenerL1 = dataBroker.registerDataTreeChangeListener(NETCONF_NODE_TOPO_TREE_ID, new L1());
 252         listenerL2 = dataBroker.registerDataTreeChangeListener(NETCONF_NODE_TOPO_TREE_ID, new L2());
 253
 254         this.initializationSuccessful = true;
 255
 256         LOG.info("Session Initiated end. Initialization done {}", initializationSuccessful);
 257
 258     }
 259
 260     /** Blueprint destroy-method method */
 261     public void destroy() {
 262         close();
 263     }
 264
 265     public DomContext getDomContext() {
 266         return Objects.requireNonNull(domContext, "Initialization not completed for domContext" );
 267     }
 268
 269     public NetconfnodeStateServiceRpcApiImpl getNetconfnodeStateServiceRpcApiImpl() {
 270         return Objects.requireNonNull(rpcApiService, "Initialization not completed for rpcApiService" );
 271     }
 272
 273     @Override
 274     public GetStatusOutputBuilder getStatus(GetStatusInput input) {
 275         return new GetStatusOutputBuilder();
 276     }
 277
 278     @Override
 279     public <L extends NetconfNodeConnectListener> @NonNull ListenerRegistration<L> registerNetconfNodeConnectListener(
 280             final @NonNull L netconfNodeConnectListener) {
 281         LOG.info("Register connect listener {}", netconfNodeConnectListener.getClass().getName());
 282         netconfNodeConnectListenerList.add(netconfNodeConnectListener);
 283
 284         return new ListenerRegistration<L>() {
 285             @Override
 286             public @NonNull L getInstance() {
 287                 return netconfNodeConnectListener;
 288             }
 289
 290             @Override
 291             public void close() {
 292                 LOG.info("Remove connect listener {}", netconfNodeConnectListener);
 293                 netconfNodeConnectListenerList.remove(netconfNodeConnectListener);
 294             }
 295         };
 296     }
 297
 298     @Override
 299     public <L extends NetconfNodeStateListener> @NonNull ListenerRegistration<L> registerNetconfNodeStateListener(
 300             @NonNull L netconfNodeStateListener) {
 301         LOG.info("Register state listener {}", netconfNodeStateListener.getClass().getName());
 302         netconfNodeStateListenerList.add(netconfNodeStateListener);
 303
 304         return new ListenerRegistration<L>() {
 305             @Override
 306             public @NonNull L getInstance() {
 307                 return netconfNodeStateListener;
 308             }
 309
 310             @Override
 311             public void close() {
 312                 LOG.info("Remove state listener {}", netconfNodeStateListener);
 313                 netconfNodeStateListenerList.remove(netconfNodeStateListener);
 314             }
 315         };
 316     }
 317
 318     @Override
 319     public <L extends VesNotificationListener> @NonNull ListenerRegistration<L> registerVesNotifications(
 320             @NonNull L vesNotificationListener) {
 321         LOG.info("Register Ves notification listener {}", vesNotificationListener.getClass().getName());
 322         vesNotificationListenerList.add(vesNotificationListener);
 323
 324         return new ListenerRegistration<L>() {
 325             @Override
 326             public @NonNull L getInstance() {
 327                 return vesNotificationListener;
 328             }
 329
 330             @Override
 331             public void close() {
 332                 LOG.info("Remove Ves notification listener {}", vesNotificationListener);
 333                 vesNotificationListenerList.remove(vesNotificationListener);
 334             }
 335         };
 336     }
 337
 338     @Override
 339     public void close() {
 340         LOG.info("Closing start ...");
 341         try {
 342             close(rpcApiService, listenerL1, listenerL2);
 343         } catch (Exception e) {
 344             LOG.debug("Closing", e);
 345         }
 346         LOG.info("Closing done");
 347     }
 348
 349     /**
 350      * Used to close all Services, that should support AutoCloseable Pattern
 351      *
 352      * @param toClose
 353      * @throws Exception
 354      */
 355     private void close(AutoCloseable... toCloseList) throws Exception {
 356         for (AutoCloseable element : toCloseList) {
 357             if (element != null) {
 358                 element.close();
 359             }
 360         }
 361         this.configFileRepresentation.unregisterConfigChangedListener(this);
 362     }
 363
 364     /**
 365      * Indication if init() of this bundle successfully done.
 366      *
 367      * @return true if init() was successful. False if not done or not successful.
 368      */
 369     public boolean isInitializationSuccessful() {
 370         return this.initializationSuccessful;
 371     }
 372
 373     /*-------------------------------------------------------------------------------------------
 374      * Functions for interface DeviceManagerService
 375      */
 376
 377     /**
 378      * For each mounted device a mountpoint is created and this listener is called. Mountpoint was created or existing.
 379      * Managed device is now fully connected to node/mountpoint.
 380      *
 381      * @param nNodeId id of the mountpoint
 382      * @param netconfNode mountpoint contents
 383      */
 384     private void enterConnectedState(NodeId nNodeId, NetconfNode netconfNode) {
 385
 386         String mountPointNodeName = nNodeId.getValue();
 387         LOG.info("Access connected state for mountpoint {}", mountPointNodeName);
 388
 389         boolean preConditionMissing = false;
 390         if (mountPointService == null) {
 391             preConditionMissing = true;
 392             LOG.warn("No mountservice available.");
 393         }
 394         if (!initializationSuccessful) {
 395             preConditionMissing = true;
 396             LOG.warn("Devicemanager initialization still pending.");
 397         }
 398         if (preConditionMissing) {
 399             return;
 400         }
 401
 402         boolean isNetconfNodeMaster = isNetconfNodeMaster(netconfNode);
 403         LOG.info("isNetconfNodeMaster indication {} for mountpoint {}", isNetconfNodeMaster, mountPointNodeName);
 404         if (isNetconfNodeMaster) {
 405             NetconfAccessor acessor = accessorManager.getAccessor(nNodeId, netconfNode);
 406                 /*
 407                  * --> Call Listers for onConnect() Indication
 408                    for (all)
 409                  */
 410                 netconfNodeConnectListenerList.forEach(item -> {
 411                     try {
 412                         item.onEnterConnected(acessor);
 413                     } catch (Exception e) {
 414                         LOG.info("Exception during onEnterConnected listener call", e);
 415                     }
 416                 });
 417
 418                 LOG.info("Connect indication forwarded for {}", mountPointNodeName);
 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             @NonNull
 655             String masterNodeName =
 656                     ccs == null || ccs.getNetconfMasterNode() == null ? "null" : ccs.getNetconfMasterNode();
 657             LOG.debug("sdnMasterNode=" + masterNodeName + " and sdnMyNode=" + clusterName);
 658             if (!masterNodeName.equals(clusterName)) {
 659                 LOG.debug("netconf change but me is not master for this node");
 660                 return false;
 661             }
 662         }
 663         return true;
 664     }
 665
 666
 667
 668     @Override
 669     public void onConfigChanged() {
 670         this.handleDataTreeAsync = this.config.handleAsync();
 671
 672     }
 673 }