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.FluentFuture;
  11 import java.util.ArrayList;
  12 import java.util.List;
  13 import java.util.Set;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import java.util.concurrent.ExecutionException;
  16 import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
  17 import org.opendaylight.mdsal.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                                         final 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     @Override
  62     void activateNotificationSource() {
  63         if (!isActive()) {
  64             LOG.info("Stream {} is not active on node {}. Will subscribe.", this.getStreamName(), this.nodeId);
  65             final FluentFuture<DOMRpcResult> result = mountPoint.invokeCreateSubscription(stream);
  66             try {
  67                 result.get();
  68                 setActive(true);
  69             } catch (InterruptedException | ExecutionException e) {
  70                 LOG.warn("Can not subscribe stream {} on node {}", this.getSourceName(), this.nodeId, e);
  71                 setActive(false);
  72             }
  73         } else {
  74             LOG.info("Stream {} is now active on node {}", this.getStreamName(), this.nodeId);
  75         }
  76     }
  77
  78     /**
  79      * Subscribes to notification stream associated with this registration. If replay is supported, notifications
  80      * from last
  81      * received event time will be requested.
  82      */
  83     @Override
  84     void reActivateNotificationSource() {
  85         if (isActive()) {
  86             LOG.info("Stream {} is reactivating on node {}.", this.getStreamName(), this.nodeId);
  87             final FluentFuture<DOMRpcResult> result = mountPoint.invokeCreateSubscription(stream, getLastEventTime());
  88             try {
  89                 result.get();
  90                 setActive(true);
  91             } catch (InterruptedException | ExecutionException e) {
  92                 LOG.warn("Can not resubscribe stream {} on node {}", this.getSourceName(), this.nodeId, e);
  93                 setActive(false);
  94             }
  95         }
  96     }
  97
  98     @Override
  99     void deActivateNotificationSource() {
 100         // no operations need
 101     }
 102
 103     private void closeStream() {
 104         if (isActive()) {
 105             for (ListenerRegistration<DOMNotificationListener> reg : notificationRegistrationMap.values()) {
 106                 reg.close();
 107             }
 108             notificationRegistrationMap.clear();
 109             notificationTopicMap.clear();
 110             setActive(false);
 111         }
 112     }
 113
 114     private String getStreamName() {
 115         return getSourceName();
 116     }
 117
 118     @Override
 119     boolean registerNotificationTopic(final SchemaPath notificationPath, final TopicId topicId) {
 120         if (!checkNotificationPath(notificationPath)) {
 121             LOG.debug("Bad SchemaPath for notification try to register");
 122             return false;
 123         }
 124
 125         activateNotificationSource();
 126         if (!isActive()) {
 127             LOG.warn("Stream {} is not active, listener for notification {} is not registered.", getStreamName(),
 128                     notificationPath.toString());
 129             return false;
 130         }
 131
 132         ListenerRegistration<DOMNotificationListener> registration =
 133                 mountPoint.registerNotificationListener(netconfEventSource, notificationPath);
 134         notificationRegistrationMap.put(notificationPath, registration);
 135         Set<TopicId> topicIds = getTopicsForNotification(notificationPath);
 136         topicIds.add(topicId);
 137
 138         notificationTopicMap.put(notificationPath, topicIds);
 139         return true;
 140     }
 141
 142     @Override
 143     synchronized void unRegisterNotificationTopic(final TopicId topicId) {
 144         List<SchemaPath> notificationPathToRemove = new ArrayList<>();
 145         for (SchemaPath notifKey : notificationTopicMap.keySet()) {
 146             Set<TopicId> topicList = notificationTopicMap.get(notifKey);
 147             if (topicList != null) {
 148                 topicList.remove(topicId);
 149                 if (topicList.isEmpty()) {
 150                     notificationPathToRemove.add(notifKey);
 151                 }
 152             }
 153         }
 154         for (SchemaPath notifKey : notificationPathToRemove) {
 155             notificationTopicMap.remove(notifKey);
 156             ListenerRegistration<DOMNotificationListener> reg = notificationRegistrationMap.remove(notifKey);
 157             if (reg != null) {
 158                 reg.close();
 159             }
 160         }
 161     }
 162
 163     @Override
 164     public void close() {
 165         closeStream();
 166     }
 167
 168 }