netconf/mdsal-netconf-notification/src/main/java/org/opendaylight/netconf/mdsal/notification/impl/NetconfNotificationManager.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.mdsal.notification.impl;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Preconditions.checkState;
  12 import static java.util.Objects.requireNonNull;
  13
  14 import com.google.common.collect.HashMultimap;
  15 import com.google.common.collect.HashMultiset;
  16 import com.google.common.collect.Maps;
  17 import com.google.common.collect.Multimap;
  18 import com.google.common.collect.Multiset;
  19 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  20 import java.util.ArrayList;
  21 import java.util.HashMap;
  22 import java.util.HashSet;
  23 import java.util.Map;
  24 import java.util.Set;
  25 import javax.inject.Inject;
  26 import javax.inject.Singleton;
  27 import org.checkerframework.checker.lock.qual.GuardedBy;
  28 import org.opendaylight.netconf.mdsal.notification.impl.ops.NotificationsTransformUtil;
  29 import org.opendaylight.netconf.notifications.BaseNotificationPublisherRegistration;
  30 import org.opendaylight.netconf.notifications.NetconfNotification;
  31 import org.opendaylight.netconf.notifications.NetconfNotificationCollector;
  32 import org.opendaylight.netconf.notifications.NetconfNotificationListener;
  33 import org.opendaylight.netconf.notifications.NetconfNotificationRegistry;
  34 import org.opendaylight.netconf.notifications.NotificationListenerRegistration;
  35 import org.opendaylight.netconf.notifications.NotificationPublisherRegistration;
  36 import org.opendaylight.netconf.notifications.NotificationRegistration;
  37 import org.opendaylight.netconf.notifications.YangLibraryPublisherRegistration;
  38 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
  39 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.Streams;
  40 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.StreamsBuilder;
  41 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.Stream;
  42 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.StreamBuilder;
  43 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.StreamKey;
  44 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChange;
  45 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfSessionEnd;
  46 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfSessionStart;
  47 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.library.rev190104.YangLibraryChange;
  48 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.library.rev190104.YangLibraryUpdate;
  49 import org.opendaylight.yangtools.yang.binding.Notification;
  50 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 /**
  55  *  A thread-safe implementation NetconfNotificationRegistry.
  56  */
  57 @Singleton
  58 public class NetconfNotificationManager implements NetconfNotificationCollector, NetconfNotificationRegistry,
  59         NetconfNotificationListener, AutoCloseable {
  60
  61     public static final StreamNameType BASE_STREAM_NAME = new StreamNameType("NETCONF");
  62     public static final Stream BASE_NETCONF_STREAM = new StreamBuilder()
  63                 .setName(BASE_STREAM_NAME)
  64                 .withKey(new StreamKey(BASE_STREAM_NAME))
  65                 .setReplaySupport(false)
  66                 .setDescription("Default Event Stream")
  67                 .build();
  68
  69     private static final Logger LOG = LoggerFactory.getLogger(NetconfNotificationManager.class);
  70
  71     // TODO excessive synchronization provides thread safety but is most likely not optimal
  72     // (combination of concurrent collections might improve performance)
  73     // And also calling callbacks from a synchronized block is dangerous
  74     // since the listeners/publishers can block the whole notification processing
  75
  76     @GuardedBy("this")
  77     private final Multimap<StreamNameType, GenericNotificationListenerReg> notificationListeners =
  78             HashMultimap.create();
  79
  80     @GuardedBy("this")
  81     private final Set<NetconfNotificationStreamListener> streamListeners = new HashSet<>();
  82
  83     @GuardedBy("this")
  84     private final Map<StreamNameType, Stream> streamMetadata = new HashMap<>();
  85
  86     @GuardedBy("this")
  87     private final Multiset<StreamNameType> availableStreams = HashMultiset.create();
  88
  89     @GuardedBy("this")
  90     private final Set<GenericNotificationPublisherReg> notificationPublishers = new HashSet<>();
  91     private final NotificationsTransformUtil transformUtil;
  92
  93     @Inject
  94     public NetconfNotificationManager(final NotificationsTransformUtil transformUtil) {
  95         this.transformUtil = requireNonNull(transformUtil);
  96     }
  97
  98     @Override
  99     public synchronized void onNotification(final StreamNameType stream, final NetconfNotification notification) {
 100         LOG.debug("Notification of type {} detected", stream);
 101         if (LOG.isTraceEnabled()) {
 102             LOG.debug("Notification of type {} detected: {}", stream, notification);
 103         }
 104
 105         for (final GenericNotificationListenerReg listenerReg : notificationListeners.get(stream)) {
 106             listenerReg.getListener().onNotification(stream, notification);
 107         }
 108     }
 109
 110     @Override
 111     public synchronized NotificationListenerRegistration registerNotificationListener(
 112             final StreamNameType stream,
 113             final NetconfNotificationListener listener) {
 114         requireNonNull(stream);
 115         requireNonNull(listener);
 116
 117         LOG.trace("Notification listener registered for stream: {}", stream);
 118
 119         final GenericNotificationListenerReg reg = new GenericNotificationListenerReg(listener, stream) {
 120             @Override
 121             public void close() {
 122                 synchronized (NetconfNotificationManager.this) {
 123                     LOG.trace("Notification listener unregistered for stream: {}", stream);
 124                     super.close();
 125                 }
 126             }
 127         };
 128
 129         notificationListeners.put(stream, reg);
 130         return reg;
 131     }
 132
 133     @Override
 134     public synchronized Streams getNotificationPublishers() {
 135         return new StreamsBuilder().setStream(Maps.uniqueIndex(streamMetadata.values(), Stream::key)).build();
 136     }
 137
 138     @Override
 139     public synchronized boolean isStreamAvailable(final StreamNameType streamNameType) {
 140         return availableStreams.contains(streamNameType);
 141     }
 142
 143     @Override
 144     public synchronized NotificationRegistration registerStreamListener(
 145             final NetconfNotificationStreamListener listener) {
 146         streamListeners.add(listener);
 147
 148         // Notify about all already available
 149         for (final Stream availableStream : streamMetadata.values()) {
 150             listener.onStreamRegistered(availableStream);
 151         }
 152
 153         return () -> {
 154             synchronized (NetconfNotificationManager.this) {
 155                 streamListeners.remove(listener);
 156             }
 157         };
 158     }
 159
 160     @Override
 161     public synchronized void close() {
 162         // Unregister all listeners
 163         // Use new list to avoid ConcurrentModificationException
 164         for (final GenericNotificationListenerReg listenerReg : new ArrayList<>(notificationListeners.values())) {
 165             listenerReg.close();
 166         }
 167         notificationListeners.clear();
 168
 169         // Unregister all publishers
 170         // Use new list to avoid ConcurrentModificationException
 171         for (final GenericNotificationPublisherReg notificationPublisher : new ArrayList<>(notificationPublishers)) {
 172             notificationPublisher.close();
 173         }
 174         notificationPublishers.clear();
 175
 176         // Clear stream Listeners
 177         streamListeners.clear();
 178     }
 179
 180     @Override
 181     public synchronized NotificationPublisherRegistration registerNotificationPublisher(final Stream stream) {
 182         final StreamNameType streamName = requireNonNull(stream).getName();
 183
 184         LOG.debug("Notification publisher registered for stream: {}", streamName);
 185         if (LOG.isTraceEnabled()) {
 186             LOG.trace("Notification publisher registered for stream: {}", stream);
 187         }
 188
 189         if (streamMetadata.containsKey(streamName)) {
 190             LOG.warn("Notification stream {} already registered as: {}. Will be reused", streamName,
 191                     streamMetadata.get(streamName));
 192         } else {
 193             streamMetadata.put(streamName, stream);
 194         }
 195
 196         availableStreams.add(streamName);
 197
 198         final GenericNotificationPublisherReg reg = new GenericNotificationPublisherReg(this, streamName) {
 199             @Override
 200             public void close() {
 201                 synchronized (NetconfNotificationManager.this) {
 202                     super.close();
 203                 }
 204             }
 205         };
 206
 207         notificationPublishers.add(reg);
 208
 209         notifyStreamAdded(stream);
 210         return reg;
 211     }
 212
 213     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 214             justification = "https://github.com/spotbugs/spotbugs/issues/811")
 215     private void unregisterNotificationPublisher(
 216             final StreamNameType streamName,
 217             final GenericNotificationPublisherReg genericNotificationPublisherReg) {
 218         availableStreams.remove(streamName);
 219         notificationPublishers.remove(genericNotificationPublisherReg);
 220
 221         LOG.debug("Notification publisher unregistered for stream: {}", streamName);
 222
 223         // Notify stream listeners if all publishers are gone and also clear metadata for stream
 224         if (!isStreamAvailable(streamName)) {
 225             LOG.debug("Notification stream: {} became unavailable", streamName);
 226             streamMetadata.remove(streamName);
 227             notifyStreamRemoved(streamName);
 228         }
 229     }
 230
 231     private synchronized void notifyStreamAdded(final Stream stream) {
 232         for (final NetconfNotificationStreamListener streamListener : streamListeners) {
 233             streamListener.onStreamRegistered(stream);
 234         }
 235     }
 236
 237     private synchronized void notifyStreamRemoved(final StreamNameType stream) {
 238         for (final NetconfNotificationStreamListener streamListener : streamListeners) {
 239             streamListener.onStreamUnregistered(stream);
 240         }
 241     }
 242
 243     @Override
 244     public BaseNotificationPublisherRegistration registerBaseNotificationPublisher() {
 245         final NotificationPublisherRegistration notificationPublisherRegistration =
 246                 registerNotificationPublisher(BASE_NETCONF_STREAM);
 247         return new BaseNotificationPublisherReg(transformUtil, notificationPublisherRegistration);
 248     }
 249
 250     @Override
 251     public YangLibraryPublisherRegistration registerYangLibraryPublisher() {
 252         final NotificationPublisherRegistration notificationPublisherRegistration =
 253                 registerNotificationPublisher(BASE_NETCONF_STREAM);
 254         return new YangLibraryPublisherReg(transformUtil, notificationPublisherRegistration);
 255     }
 256
 257     private static class GenericNotificationPublisherReg implements NotificationPublisherRegistration {
 258         private NetconfNotificationManager baseListener;
 259         private final StreamNameType registeredStream;
 260
 261         GenericNotificationPublisherReg(final NetconfNotificationManager baseListener,
 262                                         final StreamNameType registeredStream) {
 263             this.baseListener = baseListener;
 264             this.registeredStream = registeredStream;
 265         }
 266
 267         @Override
 268         public void close() {
 269             baseListener.unregisterNotificationPublisher(registeredStream, this);
 270             baseListener = null;
 271         }
 272
 273         @Override
 274         public void onNotification(final StreamNameType stream, final NetconfNotification notification) {
 275             checkState(baseListener != null, "Already closed");
 276             checkArgument(stream.equals(registeredStream));
 277             baseListener.onNotification(stream, notification);
 278         }
 279     }
 280
 281     private static class BaseNotificationPublisherReg implements BaseNotificationPublisherRegistration {
 282
 283         static final SchemaPath CAPABILITY_CHANGE_SCHEMA_PATH = SchemaPath.create(true, NetconfCapabilityChange.QNAME);
 284         static final SchemaPath SESSION_START_PATH = SchemaPath.create(true, NetconfSessionStart.QNAME);
 285         static final SchemaPath SESSION_END_PATH = SchemaPath.create(true, NetconfSessionEnd.QNAME);
 286
 287         private final NotificationPublisherRegistration baseRegistration;
 288         private final NotificationsTransformUtil transformUtil;
 289
 290         BaseNotificationPublisherReg(final NotificationsTransformUtil transformUtil,
 291                 final NotificationPublisherRegistration baseRegistration) {
 292             this.transformUtil = requireNonNull(transformUtil);
 293             this.baseRegistration = baseRegistration;
 294         }
 295
 296         @Override
 297         public void close() {
 298             baseRegistration.close();
 299         }
 300
 301         private NetconfNotification serializeNotification(final Notification notification, final SchemaPath path) {
 302             return transformUtil.transform(notification, path);
 303         }
 304
 305         @Override
 306         public void onCapabilityChanged(final NetconfCapabilityChange capabilityChange) {
 307             baseRegistration.onNotification(BASE_STREAM_NAME,
 308                     serializeNotification(capabilityChange, CAPABILITY_CHANGE_SCHEMA_PATH));
 309         }
 310
 311         @Override
 312         public void onSessionStarted(final NetconfSessionStart start) {
 313             baseRegistration.onNotification(BASE_STREAM_NAME, serializeNotification(start, SESSION_START_PATH));
 314         }
 315
 316         @Override
 317         public void onSessionEnded(final NetconfSessionEnd end) {
 318             baseRegistration.onNotification(BASE_STREAM_NAME, serializeNotification(end, SESSION_END_PATH));
 319         }
 320     }
 321
 322     private static class YangLibraryPublisherReg implements YangLibraryPublisherRegistration {
 323         static final SchemaPath YANG_LIBRARY_CHANGE_PATH = SchemaPath.create(true, YangLibraryChange.QNAME);
 324         static final SchemaPath YANG_LIBRARY_UPDATE_PATH = SchemaPath.create(true, YangLibraryUpdate.QNAME);
 325
 326         private final NotificationPublisherRegistration baseRegistration;
 327         private final NotificationsTransformUtil transformUtil;
 328
 329         YangLibraryPublisherReg(final NotificationsTransformUtil transformUtil,
 330                 final NotificationPublisherRegistration baseRegistration) {
 331             this.transformUtil = requireNonNull(transformUtil);
 332             this.baseRegistration = baseRegistration;
 333         }
 334
 335         @Override
 336         @Deprecated
 337         public void onYangLibraryChange(final YangLibraryChange yangLibraryChange) {
 338             baseRegistration.onNotification(BASE_STREAM_NAME,
 339                 transformUtil.transform(yangLibraryChange, YANG_LIBRARY_CHANGE_PATH));
 340         }
 341
 342         @Override
 343         public void onYangLibraryUpdate(final YangLibraryUpdate yangLibraryUpdate) {
 344             baseRegistration.onNotification(BASE_STREAM_NAME,
 345                     transformUtil.transform(yangLibraryUpdate, YANG_LIBRARY_UPDATE_PATH));
 346         }
 347
 348         @Override
 349         public void close() {
 350             baseRegistration.close();
 351         }
 352     }
 353
 354     private class GenericNotificationListenerReg implements NotificationListenerRegistration {
 355         private final NetconfNotificationListener listener;
 356         private final StreamNameType listenedStream;
 357
 358         GenericNotificationListenerReg(final NetconfNotificationListener listener,
 359                                        final StreamNameType listenedStream) {
 360             this.listener = listener;
 361             this.listenedStream = listenedStream;
 362         }
 363
 364         public NetconfNotificationListener getListener() {
 365             return listener;
 366         }
 367
 368         @Override
 369         public void close() {
 370             notificationListeners.remove(listenedStream, this);
 371         }
 372     }
 373 }