2 * Copyright (c) 2013 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
9 package org.opendaylight.controller.messagebus.app.impl;
11 import com.google.common.base.Preconditions;
12 import java.util.ArrayList;
13 import java.util.List;
16 import java.util.concurrent.Future;
17 import java.util.regex.Pattern;
18 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
19 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
20 import org.opendaylight.controller.mdsal.MdSAL;
21 import org.opendaylight.controller.sal.core.api.notify.NotificationListener;
22 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.NotificationPattern;
23 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicNotification;
24 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.EventSourceService;
25 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicInput;
26 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutput;
27 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
28 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInputBuilder;
29 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.NotificationsService;
30 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.inventory.rev140108.NetconfNode;
32 import org.opendaylight.yangtools.concepts.ListenerRegistration;
33 import org.opendaylight.yangtools.yang.binding.DataObject;
34 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
35 import org.opendaylight.yangtools.yang.common.QName;
36 import org.opendaylight.yangtools.yang.common.RpcResult;
37 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
38 import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode;
39 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
40 import org.slf4j.Logger;
41 import org.slf4j.LoggerFactory;
43 public class NetconfEventSource implements EventSourceService, NotificationListener, DataChangeListener {
44 private static final Logger LOGGER = LoggerFactory.getLogger(NetconfEventSource.class);
46 private final MdSAL mdSal;
47 private final String nodeId;
49 private final List<String> activeStreams = new ArrayList<>();
51 private final Map<String, String> urnPrefixToStreamMap;
53 public NetconfEventSource(final MdSAL mdSal, final String nodeId, final Map<String, String> streamMap) {
54 Preconditions.checkNotNull(mdSal);
55 Preconditions.checkNotNull(nodeId);
59 this.urnPrefixToStreamMap = streamMap;
61 LOGGER.info("NetconfEventSource [{}] created.", nodeId);
65 public Future<RpcResult<JoinTopicOutput>> joinTopic(final JoinTopicInput input) {
66 final NotificationPattern notificationPattern = input.getNotificationPattern();
68 // FIXME: default language should already be regex
69 final String regex = Util.wildcardToRegex(notificationPattern.getValue());
71 final Pattern pattern = Pattern.compile(regex);
72 List<QName> matchingNotifications = Util.expandQname(availableNotifications(), pattern);
73 registerNotificationListener(matchingNotifications);
77 private List<QName> availableNotifications() {
78 // FIXME: use SchemaContextListener to get changes asynchronously
79 Set<NotificationDefinition> availableNotifications = mdSal.getSchemaContext(nodeId).getNotifications();
80 List<QName> qNs = new ArrayList<>(availableNotifications.size());
81 for (NotificationDefinition nd : availableNotifications) {
82 qNs.add(nd.getQName());
88 private void registerNotificationListener(final List<QName> notificationsToSubscribe) {
89 for (QName qName : notificationsToSubscribe) {
90 startSubscription(qName);
91 // FIXME: do not lose this registration
92 final ListenerRegistration<NotificationListener> reg = mdSal.addNotificationListener(nodeId, qName, this);
96 private synchronized void startSubscription(final QName qName) {
97 String streamName = resolveStream(qName);
99 if (streamIsActive(streamName) == false) {
100 LOGGER.info("Stream {} is not active on node {}. Will subscribe.", streamName, nodeId);
101 startSubscription(streamName);
105 private synchronized void resubscribeToActiveStreams() {
106 for (String streamName : activeStreams) {
107 startSubscription(streamName);
111 private synchronized void startSubscription(final String streamName) {
112 CreateSubscriptionInput subscriptionInput = getSubscriptionInput(streamName);
113 mdSal.getRpcService(nodeId, NotificationsService.class).createSubscription(subscriptionInput);
114 activeStreams.add(streamName);
117 private static CreateSubscriptionInput getSubscriptionInput(final String streamName) {
118 CreateSubscriptionInputBuilder csib = new CreateSubscriptionInputBuilder();
119 csib.setStream(new StreamNameType(streamName));
123 private String resolveStream(final QName qName) {
124 String streamName = null;
126 for (Map.Entry<String, String> entry : urnPrefixToStreamMap.entrySet()) {
127 String nameSpace = qName.getNamespace().toString();
128 String urnPrefix = entry.getKey();
129 if( nameSpace.startsWith(urnPrefix) ) {
130 streamName = entry.getValue();
138 private boolean streamIsActive(final String streamName) {
139 return activeStreams.contains(streamName);
143 @Override public Set<QName> getSupportedNotifications() {
148 public void onNotification(final CompositeNode notification) {
149 LOGGER.info("NetconfEventSource {} received notification {}. Will publish to MD-SAL.", nodeId, notification);
150 ImmutableCompositeNode payload = ImmutableCompositeNode.builder()
151 .setQName(QName.create(TopicNotification.QNAME, "payload"))
152 .add(notification).toInstance();
153 ImmutableCompositeNode icn = ImmutableCompositeNode.builder()
154 .setQName(TopicNotification.QNAME)
156 .addLeaf("event-source", nodeId)
159 mdSal.publishNotification(icn);
163 public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> change) {
164 boolean wasConnected = false;
165 boolean nowConnected = false;
167 for (Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getOriginalData().entrySet()) {
168 if ( isNetconfNode(changeEntry) ) {
169 NetconfNode nn = (NetconfNode)changeEntry.getValue();
170 wasConnected = nn.isConnected();
174 for (Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getUpdatedData().entrySet()) {
175 if ( isNetconfNode(changeEntry) ) {
176 NetconfNode nn = (NetconfNode)changeEntry.getValue();
177 nowConnected = nn.isConnected();
181 if (wasConnected == false && nowConnected == true) {
182 resubscribeToActiveStreams();
186 private static boolean isNetconfNode(final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry ) {
187 return NetconfNode.class.equals(changeEntry.getKey().getTargetType());