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