2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.netconf.mdsal.notification.impl;
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;
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 java.util.ArrayList;
20 import java.util.HashMap;
21 import java.util.HashSet;
24 import javax.annotation.PreDestroy;
25 import javax.inject.Inject;
26 import javax.inject.Singleton;
27 import org.checkerframework.checker.lock.qual.GuardedBy;
28 import org.opendaylight.mdsal.binding.dom.codec.spi.BindingDOMCodecFactory;
29 import org.opendaylight.mdsal.binding.runtime.api.BindingRuntimeGenerator;
30 import org.opendaylight.netconf.notifications.BaseNotificationPublisherRegistration;
31 import org.opendaylight.netconf.notifications.NetconfNotification;
32 import org.opendaylight.netconf.notifications.NetconfNotificationCollector;
33 import org.opendaylight.netconf.notifications.NetconfNotificationListener;
34 import org.opendaylight.netconf.notifications.NetconfNotificationRegistry;
35 import org.opendaylight.netconf.notifications.NotificationListenerRegistration;
36 import org.opendaylight.netconf.notifications.NotificationPublisherRegistration;
37 import org.opendaylight.netconf.notifications.NotificationRegistration;
38 import org.opendaylight.netconf.notifications.YangLibraryPublisherRegistration;
39 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
40 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.Streams;
41 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.StreamsBuilder;
42 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.Stream;
43 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.StreamBuilder;
44 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.StreamKey;
45 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChange;
46 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfSessionEnd;
47 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfSessionStart;
48 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.library.rev190104.YangLibraryChange;
49 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.library.rev190104.YangLibraryUpdate;
50 import org.opendaylight.yangtools.yang.binding.Notification;
51 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
52 import org.opendaylight.yangtools.yang.parser.api.YangParserException;
53 import org.opendaylight.yangtools.yang.parser.api.YangParserFactory;
54 import org.osgi.service.component.annotations.Activate;
55 import org.osgi.service.component.annotations.Component;
56 import org.osgi.service.component.annotations.Deactivate;
57 import org.osgi.service.component.annotations.Reference;
58 import org.slf4j.Logger;
59 import org.slf4j.LoggerFactory;
62 * A thread-safe implementation NetconfNotificationRegistry.
65 @Component(service = { NetconfNotificationCollector.class, NetconfNotificationRegistry.class }, immediate = true,
66 property = "type=netconf-notification-manager")
67 public final class NetconfNotificationManager implements NetconfNotificationCollector, NetconfNotificationRegistry,
68 NetconfNotificationListener, AutoCloseable {
69 public static final StreamNameType BASE_STREAM_NAME = new StreamNameType("NETCONF");
70 public static final Stream BASE_NETCONF_STREAM = new StreamBuilder()
71 .setName(BASE_STREAM_NAME)
72 .withKey(new StreamKey(BASE_STREAM_NAME))
73 .setReplaySupport(false)
74 .setDescription("Default Event Stream")
77 private static final Logger LOG = LoggerFactory.getLogger(NetconfNotificationManager.class);
79 // TODO excessive synchronization provides thread safety but is most likely not optimal
80 // (combination of concurrent collections might improve performance)
81 // And also calling callbacks from a synchronized block is dangerous
82 // since the listeners/publishers can block the whole notification processing
85 private final Multimap<StreamNameType, GenericNotificationListenerReg> notificationListeners =
86 HashMultimap.create();
89 private final Set<NetconfNotificationStreamListener> streamListeners = new HashSet<>();
92 private final Map<StreamNameType, Stream> streamMetadata = new HashMap<>();
95 private final Multiset<StreamNameType> availableStreams = HashMultiset.create();
98 private final Set<GenericNotificationPublisherReg> notificationPublishers = new HashSet<>();
99 private final NotificationsTransformUtil transformUtil;
103 public NetconfNotificationManager(@Reference final YangParserFactory parserFactory,
104 @Reference final BindingRuntimeGenerator generator, @Reference final BindingDOMCodecFactory codecFactory)
105 throws YangParserException {
106 transformUtil = new NotificationsTransformUtil(parserFactory, generator, codecFactory);
112 public synchronized void close() {
113 // Unregister all listeners
114 // Use new list to avoid ConcurrentModificationException
115 for (final GenericNotificationListenerReg listenerReg : new ArrayList<>(notificationListeners.values())) {
118 notificationListeners.clear();
120 // Unregister all publishers
121 // Use new list to avoid ConcurrentModificationException
122 for (final GenericNotificationPublisherReg notificationPublisher : new ArrayList<>(notificationPublishers)) {
123 notificationPublisher.close();
125 notificationPublishers.clear();
127 // Clear stream Listeners
128 streamListeners.clear();
132 public synchronized void onNotification(final StreamNameType stream, final NetconfNotification notification) {
133 LOG.debug("Notification of type {} detected", stream);
134 if (LOG.isTraceEnabled()) {
135 LOG.debug("Notification of type {} detected: {}", stream, notification);
138 for (final GenericNotificationListenerReg listenerReg : notificationListeners.get(stream)) {
139 listenerReg.getListener().onNotification(stream, notification);
144 public synchronized NotificationListenerRegistration registerNotificationListener(
145 final StreamNameType stream,
146 final NetconfNotificationListener listener) {
147 requireNonNull(stream);
148 requireNonNull(listener);
150 LOG.trace("Notification listener registered for stream: {}", stream);
152 final GenericNotificationListenerReg reg = new GenericNotificationListenerReg(listener, stream) {
154 public void close() {
155 synchronized (NetconfNotificationManager.this) {
156 LOG.trace("Notification listener unregistered for stream: {}", stream);
162 notificationListeners.put(stream, reg);
167 public synchronized Streams getNotificationPublishers() {
168 return new StreamsBuilder().setStream(Maps.uniqueIndex(streamMetadata.values(), Stream::key)).build();
172 public synchronized boolean isStreamAvailable(final StreamNameType streamNameType) {
173 return availableStreams.contains(streamNameType);
177 public synchronized NotificationRegistration registerStreamListener(
178 final NetconfNotificationStreamListener listener) {
179 streamListeners.add(listener);
181 // Notify about all already available
182 for (final Stream availableStream : streamMetadata.values()) {
183 listener.onStreamRegistered(availableStream);
187 synchronized (NetconfNotificationManager.this) {
188 streamListeners.remove(listener);
194 public synchronized NotificationPublisherRegistration registerNotificationPublisher(final Stream stream) {
195 final StreamNameType streamName = requireNonNull(stream).getName();
197 LOG.debug("Notification publisher registered for stream: {}", streamName);
198 if (LOG.isTraceEnabled()) {
199 LOG.trace("Notification publisher registered for stream: {}", stream);
202 if (streamMetadata.containsKey(streamName)) {
203 LOG.warn("Notification stream {} already registered as: {}. Will be reused", streamName,
204 streamMetadata.get(streamName));
206 streamMetadata.put(streamName, stream);
209 availableStreams.add(streamName);
211 final GenericNotificationPublisherReg reg = new GenericNotificationPublisherReg(this, streamName) {
213 public void close() {
214 synchronized (NetconfNotificationManager.this) {
220 notificationPublishers.add(reg);
222 notifyStreamAdded(stream);
226 private void unregisterNotificationPublisher(
227 final StreamNameType streamName,
228 final GenericNotificationPublisherReg genericNotificationPublisherReg) {
229 availableStreams.remove(streamName);
230 notificationPublishers.remove(genericNotificationPublisherReg);
232 LOG.debug("Notification publisher unregistered for stream: {}", streamName);
234 // Notify stream listeners if all publishers are gone and also clear metadata for stream
235 if (!isStreamAvailable(streamName)) {
236 LOG.debug("Notification stream: {} became unavailable", streamName);
237 streamMetadata.remove(streamName);
238 notifyStreamRemoved(streamName);
242 private synchronized void notifyStreamAdded(final Stream stream) {
243 for (final NetconfNotificationStreamListener streamListener : streamListeners) {
244 streamListener.onStreamRegistered(stream);
248 private synchronized void notifyStreamRemoved(final StreamNameType stream) {
249 for (final NetconfNotificationStreamListener streamListener : streamListeners) {
250 streamListener.onStreamUnregistered(stream);
255 public BaseNotificationPublisherRegistration registerBaseNotificationPublisher() {
256 final NotificationPublisherRegistration notificationPublisherRegistration =
257 registerNotificationPublisher(BASE_NETCONF_STREAM);
258 return new BaseNotificationPublisherReg(transformUtil, notificationPublisherRegistration);
262 public YangLibraryPublisherRegistration registerYangLibraryPublisher() {
263 final NotificationPublisherRegistration notificationPublisherRegistration =
264 registerNotificationPublisher(BASE_NETCONF_STREAM);
265 return new YangLibraryPublisherReg(transformUtil, notificationPublisherRegistration);
268 private static class GenericNotificationPublisherReg implements NotificationPublisherRegistration {
269 private NetconfNotificationManager baseListener;
270 private final StreamNameType registeredStream;
272 GenericNotificationPublisherReg(final NetconfNotificationManager baseListener,
273 final StreamNameType registeredStream) {
274 this.baseListener = baseListener;
275 this.registeredStream = registeredStream;
279 public void close() {
280 baseListener.unregisterNotificationPublisher(registeredStream, this);
285 public void onNotification(final StreamNameType stream, final NetconfNotification notification) {
286 checkState(baseListener != null, "Already closed");
287 checkArgument(stream.equals(registeredStream));
288 baseListener.onNotification(stream, notification);
292 private static class BaseNotificationPublisherReg implements BaseNotificationPublisherRegistration {
293 static final Absolute CAPABILITY_CHANGE_SCHEMA_PATH = Absolute.of(NetconfCapabilityChange.QNAME);
294 static final Absolute SESSION_START_PATH = Absolute.of(NetconfSessionStart.QNAME);
295 static final Absolute SESSION_END_PATH = Absolute.of(NetconfSessionEnd.QNAME);
297 private final NotificationPublisherRegistration baseRegistration;
298 private final NotificationsTransformUtil transformUtil;
300 BaseNotificationPublisherReg(final NotificationsTransformUtil transformUtil,
301 final NotificationPublisherRegistration baseRegistration) {
302 this.transformUtil = requireNonNull(transformUtil);
303 this.baseRegistration = baseRegistration;
307 public void close() {
308 baseRegistration.close();
311 private NetconfNotification serializeNotification(final Notification<?> notification, final Absolute path) {
312 return transformUtil.transform(notification, path);
316 public void onCapabilityChanged(final NetconfCapabilityChange capabilityChange) {
317 baseRegistration.onNotification(BASE_STREAM_NAME,
318 serializeNotification(capabilityChange, CAPABILITY_CHANGE_SCHEMA_PATH));
322 public void onSessionStarted(final NetconfSessionStart start) {
323 baseRegistration.onNotification(BASE_STREAM_NAME, serializeNotification(start, SESSION_START_PATH));
327 public void onSessionEnded(final NetconfSessionEnd end) {
328 baseRegistration.onNotification(BASE_STREAM_NAME, serializeNotification(end, SESSION_END_PATH));
332 private static class YangLibraryPublisherReg implements YangLibraryPublisherRegistration {
333 static final Absolute YANG_LIBRARY_CHANGE_PATH = Absolute.of(YangLibraryChange.QNAME);
334 static final Absolute YANG_LIBRARY_UPDATE_PATH = Absolute.of(YangLibraryUpdate.QNAME);
336 private final NotificationPublisherRegistration baseRegistration;
337 private final NotificationsTransformUtil transformUtil;
339 YangLibraryPublisherReg(final NotificationsTransformUtil transformUtil,
340 final NotificationPublisherRegistration baseRegistration) {
341 this.transformUtil = requireNonNull(transformUtil);
342 this.baseRegistration = baseRegistration;
347 public void onYangLibraryChange(final YangLibraryChange yangLibraryChange) {
348 baseRegistration.onNotification(BASE_STREAM_NAME,
349 transformUtil.transform(yangLibraryChange, YANG_LIBRARY_CHANGE_PATH));
353 public void onYangLibraryUpdate(final YangLibraryUpdate yangLibraryUpdate) {
354 baseRegistration.onNotification(BASE_STREAM_NAME,
355 transformUtil.transform(yangLibraryUpdate, YANG_LIBRARY_UPDATE_PATH));
359 public void close() {
360 baseRegistration.close();
364 private class GenericNotificationListenerReg implements NotificationListenerRegistration {
365 private final NetconfNotificationListener listener;
366 private final StreamNameType listenedStream;
368 GenericNotificationListenerReg(final NetconfNotificationListener listener,
369 final StreamNameType listenedStream) {
370 this.listener = listener;
371 this.listenedStream = listenedStream;
374 public NetconfNotificationListener getListener() {
379 public void close() {
380 notificationListeners.remove(listenedStream, this);