netconf/messagebus-netconf/src/main/java/org/opendaylight/netconf/messagebus/eventsources/netconf/StreamNotificationTopicRegistration.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.netconf.messagebus.eventsources.netconf;
   9
  10 import com.google.common.util.concurrent.CheckedFuture;
  11 import java.util.ArrayList;
  12 import java.util.List;
  13 import java.util.Set;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
  16 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
  17 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  18 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicId;
  19 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.Stream;
  20 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  21 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  22 import org.slf4j.Logger;
  23 import org.slf4j.LoggerFactory;
  24
  25 /**
  26  * Topic registration for notification with specified namespace from stream.
  27  */
  28 class StreamNotificationTopicRegistration extends NotificationTopicRegistration {
  29
  30     private static final Logger LOG = LoggerFactory.getLogger(StreamNotificationTopicRegistration.class);
  31
  32     private final String nodeId;
  33     private final NetconfEventSource netconfEventSource;
  34     private final NetconfEventSourceMount mountPoint;
  35     private final ConcurrentHashMap<SchemaPath, ListenerRegistration<DOMNotificationListener>>
  36             notificationRegistrationMap = new ConcurrentHashMap<>();
  37     private final Stream stream;
  38
  39     /**
  40      * Creates registration to notification stream.
  41      *
  42      * @param stream             stream
  43      * @param notificationPrefix notifications namespace
  44      * @param netconfEventSource event source
  45      */
  46     StreamNotificationTopicRegistration(final Stream stream, final String notificationPrefix,
  47                                         NetconfEventSource netconfEventSource) {
  48         super(NotificationSourceType.NetconfDeviceStream, stream.getName().getValue(), notificationPrefix);
  49         this.netconfEventSource = netconfEventSource;
  50         this.mountPoint = netconfEventSource.getMount();
  51         this.nodeId = mountPoint.getNode().getNodeId().getValue();
  52         this.stream = stream;
  53         setReplaySupported(stream.isReplaySupport());
  54         setActive(false);
  55         LOG.info("StreamNotificationTopicRegistration initialized for {}", getStreamName());
  56     }
  57
  58     /**
  59      * Subscribes to notification stream associated with this registration.
  60      */
  61     void activateNotificationSource() {
  62         if (!isActive()) {
  63             LOG.info("Stream {} is not active on node {}. Will subscribe.", this.getStreamName(), this.nodeId);
  64             final CheckedFuture<DOMRpcResult, DOMRpcException> result = mountPoint.invokeCreateSubscription(stream);
  65             try {
  66                 result.checkedGet();
  67                 setActive(true);
  68             } catch (DOMRpcException e) {
  69                 LOG.warn("Can not subscribe stream {} on node {}", this.getSourceName(), this.nodeId);
  70                 setActive(false);
  71             }
  72         } else {
  73             LOG.info("Stream {} is now active on node {}", this.getStreamName(), this.nodeId);
  74         }
  75     }
  76
  77     /**
  78      * Subscribes to notification stream associated with this registration. If replay is supported, notifications
  79      * from last
  80      * received event time will be requested.
  81      */
  82     void reActivateNotificationSource() {
  83         if (isActive()) {
  84             LOG.info("Stream {} is reactivating on node {}.", this.getStreamName(), this.nodeId);
  85             final CheckedFuture<DOMRpcResult, DOMRpcException> result;
  86             result = mountPoint.invokeCreateSubscription(stream, getLastEventTime());
  87             try {
  88                 result.checkedGet();
  89                 setActive(true);
  90             } catch (DOMRpcException e) {
  91                 LOG.warn("Can not resubscribe stream {} on node {}", this.getSourceName(), this.nodeId);
  92                 setActive(false);
  93             }
  94         }
  95     }
  96
  97     @Override
  98     void deActivateNotificationSource() {
  99         // no operations need
 100     }
 101
 102     private void closeStream() {
 103         if (isActive()) {
 104             for (ListenerRegistration<DOMNotificationListener> reg : notificationRegistrationMap.values()) {
 105                 reg.close();
 106             }
 107             notificationRegistrationMap.clear();
 108             notificationTopicMap.clear();
 109             setActive(false);
 110         }
 111     }
 112
 113     private String getStreamName() {
 114         return getSourceName();
 115     }
 116
 117     @Override
 118     boolean registerNotificationTopic(SchemaPath notificationPath, TopicId topicId) {
 119         if (!checkNotificationPath(notificationPath)) {
 120             LOG.debug("Bad SchemaPath for notification try to register");
 121             return false;
 122         }
 123
 124         activateNotificationSource();
 125         if (!isActive()) {
 126             LOG.warn("Stream {} is not active, listener for notification {} is not registered.", getStreamName(),
 127                     notificationPath.toString());
 128             return false;
 129         }
 130
 131         ListenerRegistration<DOMNotificationListener> registration =
 132                 mountPoint.registerNotificationListener(netconfEventSource, notificationPath);
 133         notificationRegistrationMap.put(notificationPath, registration);
 134         Set<TopicId> topicIds = getTopicsForNotification(notificationPath);
 135         topicIds.add(topicId);
 136
 137         notificationTopicMap.put(notificationPath, topicIds);
 138         return true;
 139     }
 140
 141     @Override
 142     synchronized void unRegisterNotificationTopic(TopicId topicId) {
 143         List<SchemaPath> notificationPathToRemove = new ArrayList<>();
 144         for (SchemaPath notifKey : notificationTopicMap.keySet()) {
 145             Set<TopicId> topicList = notificationTopicMap.get(notifKey);
 146             if (topicList != null) {
 147                 topicList.remove(topicId);
 148                 if (topicList.isEmpty()) {
 149                     notificationPathToRemove.add(notifKey);
 150                 }
 151             }
 152         }
 153         for (SchemaPath notifKey : notificationPathToRemove) {
 154             notificationTopicMap.remove(notifKey);
 155             ListenerRegistration<DOMNotificationListener> reg = notificationRegistrationMap.remove(notifKey);
 156             if (reg != null) {
 157                 reg.close();
 158             }
 159         }
 160     }
 161
 162     @Override
 163     public void close() throws Exception {
 164         closeStream();
 165     }
 166
 167 }