restconf/restconf-nb-bierman02/src/main/java/org/opendaylight/netconf/sal/streams/listeners/Notificator.java

   1 /*
   2  * Copyright (c) 2014 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.sal.streams.listeners;
   9
  10 import java.util.ArrayList;
  11 import java.util.List;
  12 import java.util.Map;
  13 import java.util.Set;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import java.util.concurrent.locks.Lock;
  16 import java.util.concurrent.locks.ReentrantLock;
  17 import org.opendaylight.netconf.sal.restconf.impl.ControllerContext;
  18 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  19 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  20 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
  21 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
  22 import org.slf4j.Logger;
  23 import org.slf4j.LoggerFactory;
  24
  25 /**
  26  * {@link Notificator} is responsible to create, remove and find
  27  * {@link ListenerAdapter} listener.
  28  */
  29 public final class Notificator {
  30
  31     private static Map<String, ListenerAdapter> dataChangeListener = new ConcurrentHashMap<>();
  32     private static Map<String, List<NotificationListenerAdapter>> notificationListenersByStreamName =
  33             new ConcurrentHashMap<>();
  34
  35     private static final Logger LOG = LoggerFactory.getLogger(Notificator.class);
  36     private static final Lock LOCK = new ReentrantLock();
  37
  38     private Notificator() {
  39     }
  40
  41     /**
  42      * Returns list of all stream names.
  43      */
  44     public static Set<String> getStreamNames() {
  45         return dataChangeListener.keySet();
  46     }
  47
  48     /**
  49      * Gets {@link ListenerAdapter} specified by stream name.
  50      *
  51      * @param streamName
  52      *            The name of the stream.
  53      * @return {@link ListenerAdapter} specified by stream name.
  54      */
  55     public static ListenerAdapter getListenerFor(final String streamName) {
  56         return dataChangeListener.get(streamName);
  57     }
  58
  59     /**
  60      * Checks if the listener specified by {@link YangInstanceIdentifier} path exist.
  61      *
  62      * @param streamName    name of the stream
  63      * @return True if the listener exist, false otherwise.
  64      */
  65     public static boolean existListenerFor(final String streamName) {
  66         return dataChangeListener.containsKey(streamName);
  67     }
  68
  69     /**
  70      * Creates new {@link ListenerAdapter} listener from
  71      * {@link YangInstanceIdentifier} path and stream name.
  72      *
  73      * @param path
  74      *            Path to data in data repository.
  75      * @param streamName
  76      *            The name of the stream.
  77      * @param outputType
  78      *             Spcific type of output for notifications - XML or JSON
  79      * @return New {@link ListenerAdapter} listener from
  80      *         {@link YangInstanceIdentifier} path and stream name.
  81      */
  82     public static ListenerAdapter createListener(final YangInstanceIdentifier path, final String streamName,
  83             final NotificationOutputType outputType, final ControllerContext controllerContext) {
  84         final ListenerAdapter listener = new ListenerAdapter(path, streamName, outputType, controllerContext);
  85         try {
  86             LOCK.lock();
  87             dataChangeListener.put(streamName, listener);
  88         } finally {
  89             LOCK.unlock();
  90         }
  91         return listener;
  92     }
  93
  94     /**
  95      * Looks for listener determined by {@link YangInstanceIdentifier} path and removes it.
  96      * Creates String representation of stream name from URI. Removes slash from URI in start and end position.
  97      *
  98      * @param uri
  99      *            URI for creation stream name.
 100      * @return String representation of stream name.
 101      */
 102     public static String createStreamNameFromUri(final String uri) {
 103         if (uri == null) {
 104             return null;
 105         }
 106         String result = uri;
 107         if (result.startsWith("/")) {
 108             result = result.substring(1);
 109         }
 110         if (result.endsWith("/")) {
 111             result = result.substring(0, result.length() - 1);
 112         }
 113         return result;
 114     }
 115
 116     /**
 117      * Removes all listeners.
 118      */
 119     @SuppressWarnings("checkstyle:IllegalCatch")
 120     public static void removeAllListeners() {
 121         for (final ListenerAdapter listener : dataChangeListener.values()) {
 122             try {
 123                 listener.close();
 124             } catch (final Exception e) {
 125                 LOG.error("Failed to close listener", e);
 126             }
 127         }
 128         try {
 129             LOCK.lock();
 130             dataChangeListener = new ConcurrentHashMap<>();
 131         } finally {
 132             LOCK.unlock();
 133         }
 134     }
 135
 136     /**
 137      * Delete {@link ListenerAdapter} listener specified in parameter.
 138      *
 139      * @param <T>
 140      *
 141      * @param listener
 142      *            ListenerAdapter
 143      */
 144     @SuppressWarnings("checkstyle:IllegalCatch")
 145     private static <T extends BaseListenerInterface> void deleteListener(final T listener) {
 146         if (listener != null) {
 147             try {
 148                 listener.close();
 149             } catch (final Exception e) {
 150                 LOG.error("Failed to close listener", e);
 151             }
 152             try {
 153                 LOCK.lock();
 154                 dataChangeListener.remove(listener.getStreamName());
 155             } finally {
 156                 LOCK.unlock();
 157             }
 158         }
 159     }
 160
 161     /**
 162      * Check if the listener specified by qnames of request exist.
 163      *
 164      * @param streamName
 165      *             name of stream
 166      * @return True if the listener exist, false otherwise.
 167      */
 168     public static boolean existNotificationListenerFor(final String streamName) {
 169         return notificationListenersByStreamName.containsKey(streamName);
 170     }
 171
 172     /**
 173      * Prepare listener for notification ({@link NotificationDefinition}).
 174      *
 175      * @param paths
 176      *             paths of notifications
 177      * @param streamName
 178      *             name of stream (generated by paths)
 179      * @param outputType
 180      *             type of output for onNotification - XML or JSON
 181      * @return List of {@link NotificationListenerAdapter} by paths
 182      */
 183     public static List<NotificationListenerAdapter> createNotificationListener(final List<Absolute> paths,
 184             final String streamName, final String outputType, final ControllerContext controllerContext) {
 185         final List<NotificationListenerAdapter> listListeners = new ArrayList<>();
 186         for (final Absolute path : paths) {
 187             final NotificationListenerAdapter listener =
 188                     new NotificationListenerAdapter(path, streamName, outputType, controllerContext);
 189             listListeners.add(listener);
 190         }
 191         try {
 192             LOCK.lock();
 193             notificationListenersByStreamName.put(streamName, listListeners);
 194         } finally {
 195             LOCK.unlock();
 196         }
 197         return listListeners;
 198     }
 199
 200     public static <T extends BaseListenerInterface> void removeListenerIfNoSubscriberExists(final T listener) {
 201         if (!listener.hasSubscribers()) {
 202             if (listener instanceof NotificationListenerAdapter) {
 203                 deleteNotificationListener(listener);
 204             } else {
 205                 deleteListener(listener);
 206             }
 207         }
 208     }
 209
 210     @SuppressWarnings("checkstyle:IllegalCatch")
 211     private static <T extends BaseListenerInterface> void deleteNotificationListener(final T listener) {
 212         if (listener != null) {
 213             try {
 214                 listener.close();
 215             } catch (final Exception e) {
 216                 LOG.error("Failed to close listener", e);
 217             }
 218             try {
 219                 LOCK.lock();
 220                 notificationListenersByStreamName.remove(listener.getStreamName());
 221             } finally {
 222                 LOCK.unlock();
 223             }
 224         }
 225     }
 226
 227     public static List<NotificationListenerAdapter> getNotificationListenerFor(final String streamName) {
 228         return notificationListenersByStreamName.get(streamName);
 229     }
 230 }