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
9 package org.opendaylight.controller.messagebus.eventsources.netconf;
11 import static com.google.common.util.concurrent.Futures.immediateFuture;
13 import java.io.IOException;
14 import java.util.ArrayList;
15 import java.util.List;
18 import java.util.concurrent.ConcurrentHashMap;
19 import java.util.concurrent.Future;
20 import java.util.regex.Pattern;
22 import javax.xml.stream.XMLStreamException;
23 import javax.xml.transform.dom.DOMResult;
24 import javax.xml.transform.dom.DOMSource;
26 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
27 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
28 import org.opendaylight.controller.md.sal.dom.api.DOMMountPoint;
29 import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
30 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
31 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
32 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
33 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
34 import org.opendaylight.controller.messagebus.app.impl.TopicDOMNotification;
35 import org.opendaylight.controller.messagebus.app.impl.Util;
36 import org.opendaylight.controller.messagebus.spi.EventSource;
37 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
38 import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
39 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.NotificationPattern;
40 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicId;
41 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicNotification;
42 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicInput;
43 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutput;
44 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutputBuilder;
45 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicStatus;
46 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.inventory.rev140108.NetconfNode;
48 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
49 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
50 import org.opendaylight.yangtools.concepts.ListenerRegistration;
51 import org.opendaylight.yangtools.yang.binding.DataObject;
52 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
53 import org.opendaylight.yangtools.yang.common.QName;
54 import org.opendaylight.yangtools.yang.common.RpcResult;
55 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
56 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
57 import org.opendaylight.yangtools.yang.data.api.schema.AnyXmlNode;
58 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
59 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
60 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
61 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
62 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
63 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
64 import org.slf4j.Logger;
65 import org.slf4j.LoggerFactory;
66 import org.w3c.dom.Document;
67 import org.w3c.dom.Element;
69 import com.google.common.base.Optional;
70 import com.google.common.base.Throwables;
72 public class NetconfEventSource implements EventSource, DOMNotificationListener, DataChangeListener {
74 private static final Logger LOG = LoggerFactory.getLogger(NetconfEventSource.class);
76 private static final NodeIdentifier TOPIC_NOTIFICATION_ARG = new NodeIdentifier(TopicNotification.QNAME);
77 private static final NodeIdentifier EVENT_SOURCE_ARG = new NodeIdentifier(QName.create(TopicNotification.QNAME, "node-id"));
78 private static final NodeIdentifier TOPIC_ID_ARG = new NodeIdentifier(QName.create(TopicNotification.QNAME, "topic-id"));
79 private static final NodeIdentifier PAYLOAD_ARG = new NodeIdentifier(QName.create(TopicNotification.QNAME, "payload"));
81 private static final NodeIdentifier STREAM_QNAME = new NodeIdentifier(QName.create(CreateSubscriptionInput.QNAME,"stream"));
82 private static final SchemaPath CREATE_SUBSCRIPTION = SchemaPath.create(true, QName.create(CreateSubscriptionInput.QNAME, "create-subscription"));
84 private final String nodeId;
85 private final Node node;
87 private final DOMMountPoint netconfMount;
88 private final DOMNotificationPublishService domPublish;
90 private final Map<String, String> urnPrefixToStreamMap;
92 private final ConcurrentHashMap<String, StreamNotificationTopicRegistration> streamNotifRegistrationMap = new ConcurrentHashMap<>();
94 public NetconfEventSource(final Node node, final Map<String, String> streamMap, final DOMMountPoint netconfMount, final DOMNotificationPublishService publishService) {
95 this.netconfMount = netconfMount;
97 this.nodeId = node.getNodeId().getValue();
98 this.urnPrefixToStreamMap = streamMap;
99 this.domPublish = publishService;
100 this.initializeStreamNotifRegistrationMap();
101 LOG.info("NetconfEventSource [{}] created.", nodeId);
104 private void initializeStreamNotifRegistrationMap(){
105 for(String streamName : this.urnPrefixToStreamMap.values()){
106 streamNotifRegistrationMap.put(streamName, new StreamNotificationTopicRegistration(streamName, this.nodeId, this.netconfMount, this));
111 public Future<RpcResult<JoinTopicOutput>> joinTopic(final JoinTopicInput input) {
113 final NotificationPattern notificationPattern = input.getNotificationPattern();
114 final List<SchemaPath> matchingNotifications = getMatchingNotifications(notificationPattern);
115 return registerTopic(input.getTopicId(),matchingNotifications);
119 private synchronized Future<RpcResult<JoinTopicOutput>> registerTopic(final TopicId topicId, final List<SchemaPath> notificationsToSubscribe){
121 JoinTopicStatus joinTopicStatus = JoinTopicStatus.Down;
122 if(notificationsToSubscribe != null && notificationsToSubscribe.isEmpty() == false){
123 final Optional<DOMNotificationService> notifyService = netconfMount.getService(DOMNotificationService.class);
124 if(notifyService.isPresent()){
125 int subscribedStreams = 0;
126 for(SchemaPath schemaNotification : notificationsToSubscribe){
127 final Optional<String> streamName = resolveStream(schemaNotification.getLastComponent());
128 if(streamName.isPresent()){
129 LOG.info("Stream {} is activating, TopicId {}", streamName.get(), topicId.getValue() );
130 StreamNotificationTopicRegistration streamReg = streamNotifRegistrationMap.get(streamName.get());
131 streamReg.activateStream();
132 for(SchemaPath notificationPath : notificationsToSubscribe){
133 LOG.info("Notification listener is registering, Notification {}, TopicId {}", notificationPath, topicId.getValue() );
134 streamReg.registerNotificationListenerTopic(notificationPath, topicId);
136 subscribedStreams = subscribedStreams + 1;
139 if(subscribedStreams > 0){
140 joinTopicStatus = JoinTopicStatus.Up;
145 final JoinTopicOutput output = new JoinTopicOutputBuilder().setStatus(joinTopicStatus).build();
146 return immediateFuture(RpcResultBuilder.success(output).build());
150 private void resubscribeToActiveStreams() {
151 for (StreamNotificationTopicRegistration streamReg : streamNotifRegistrationMap.values()){
152 streamReg.reActivateStream();
156 private Optional<String> resolveStream(final QName qName) {
157 String streamName = null;
159 for (final Map.Entry<String, String> entry : urnPrefixToStreamMap.entrySet()) {
160 final String nameSpace = qName.getNamespace().toString();
161 final String urnPrefix = entry.getKey();
162 if( nameSpace.startsWith(urnPrefix) ) {
163 streamName = entry.getValue();
167 return Optional.fromNullable(streamName);
171 public void onNotification(final DOMNotification notification) {
172 SchemaPath notificationPath = notification.getType();
173 LOG.info("Notification {} has come.",notification.getType());
174 for(StreamNotificationTopicRegistration streamReg : streamNotifRegistrationMap.values()){
175 for(TopicId topicId : streamReg.getNotificationTopicIds(notificationPath)){
176 publishNotification(notification, topicId);
177 LOG.info("Notification {} has been published for TopicId {}",notification.getType(), topicId.getValue());
182 private void publishNotification(final DOMNotification notification, TopicId topicId){
183 final ContainerNode topicNotification = Builders.containerBuilder()
184 .withNodeIdentifier(TOPIC_NOTIFICATION_ARG)
185 .withChild(ImmutableNodes.leafNode(TOPIC_ID_ARG, topicId))
186 .withChild(ImmutableNodes.leafNode(EVENT_SOURCE_ARG, nodeId))
187 .withChild(encapsulate(notification))
190 domPublish.putNotification(new TopicDOMNotification(topicNotification));
191 } catch (final InterruptedException e) {
192 throw Throwables.propagate(e);
196 private AnyXmlNode encapsulate(final DOMNotification body) {
197 // FIXME: Introduce something like AnyXmlWithNormalizedNodeData in Yangtools
198 final Document doc = XmlUtil.newDocument();
199 final Optional<String> namespace = Optional.of(PAYLOAD_ARG.getNodeType().getNamespace().toString());
200 final Element element = XmlUtil.createElement(doc , "payload", namespace);
202 final DOMResult result = new DOMResult(element);
204 final SchemaContext context = netconfMount.getSchemaContext();
205 final SchemaPath schemaPath = body.getType();
207 NetconfMessageTransformUtil.writeNormalizedNode(body.getBody(), result, schemaPath, context);
208 return Builders.anyXmlBuilder().withNodeIdentifier(PAYLOAD_ARG)
209 .withValue(new DOMSource(element))
211 } catch (IOException | XMLStreamException e) {
212 LOG.error("Unable to encapsulate notification.",e);
213 throw Throwables.propagate(e);
218 public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> change) {
219 boolean wasConnected = false;
220 boolean nowConnected = false;
222 for (final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getOriginalData().entrySet()) {
223 if ( isNetconfNode(changeEntry) ) {
224 final NetconfNode nn = (NetconfNode)changeEntry.getValue();
225 wasConnected = nn.isConnected();
229 for (final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getUpdatedData().entrySet()) {
230 if ( isNetconfNode(changeEntry) ) {
231 final NetconfNode nn = (NetconfNode)changeEntry.getValue();
232 nowConnected = nn.isConnected();
236 if (wasConnected == false && nowConnected == true) {
237 resubscribeToActiveStreams();
241 private static boolean isNetconfNode(final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry ) {
242 return NetconfNode.class.equals(changeEntry.getKey().getTargetType());
245 private List<SchemaPath> getMatchingNotifications(NotificationPattern notificationPattern){
246 // FIXME: default language should already be regex
247 final String regex = Util.wildcardToRegex(notificationPattern.getValue());
249 final Pattern pattern = Pattern.compile(regex);
250 List<SchemaPath> availableNotifications = getAvailableNotifications();
251 if(availableNotifications == null || availableNotifications.isEmpty()){
254 return Util.expandQname(availableNotifications, pattern);
258 public void close() throws Exception {
259 for(StreamNotificationTopicRegistration streamReg : streamNotifRegistrationMap.values()){
260 streamReg.deactivateStream();
265 public NodeKey getSourceNodeKey(){
266 return node.getKey();
270 public List<SchemaPath> getAvailableNotifications() {
271 // FIXME: use SchemaContextListener to get changes asynchronously
272 final Set<NotificationDefinition> availableNotifications = netconfMount.getSchemaContext().getNotifications();
273 final List<SchemaPath> qNs = new ArrayList<>(availableNotifications.size());
274 for (final NotificationDefinition nd : availableNotifications) {
275 qNs.add(nd.getPath());
280 private class StreamNotificationTopicRegistration{
282 final private String streamName;
283 final private DOMMountPoint netconfMount;
284 final private String nodeId;
285 final private NetconfEventSource notificationListener;
286 private boolean active;
288 private ConcurrentHashMap<SchemaPath, ListenerRegistration<NetconfEventSource>> notificationRegistrationMap = new ConcurrentHashMap<>();
289 private ConcurrentHashMap<SchemaPath, ArrayList<TopicId>> notificationTopicMap = new ConcurrentHashMap<>();
291 public StreamNotificationTopicRegistration(final String streamName, final String nodeId, final DOMMountPoint netconfMount, NetconfEventSource notificationListener) {
292 this.streamName = streamName;
293 this.netconfMount = netconfMount;
294 this.nodeId = nodeId;
295 this.notificationListener = notificationListener;
299 public boolean isActive() {
303 public void reActivateStream(){
305 LOG.info("Stream {} is reactivated active on node {}.", this.streamName, this.nodeId);
306 final ContainerNode input = Builders.containerBuilder().withNodeIdentifier(new NodeIdentifier(CreateSubscriptionInput.QNAME))
307 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, this.streamName))
309 netconfMount.getService(DOMRpcService.class).get().invokeRpc(CREATE_SUBSCRIPTION, input);
313 public void activateStream() {
314 if(this.isActive() == false){
315 LOG.info("Stream {} is not active on node {}. Will subscribe.", this.streamName, this.nodeId);
316 final ContainerNode input = Builders.containerBuilder().withNodeIdentifier(new NodeIdentifier(CreateSubscriptionInput.QNAME))
317 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, this.streamName))
319 netconfMount.getService(DOMRpcService.class).get().invokeRpc(CREATE_SUBSCRIPTION, input);
322 LOG.info("Stream {} is now active on node {}", this.streamName, this.nodeId);
326 public void deactivateStream() {
327 for(ListenerRegistration<NetconfEventSource> reg : notificationRegistrationMap.values()){
333 public String getStreamName() {
337 public ArrayList<TopicId> getNotificationTopicIds(SchemaPath notificationPath){
338 return notificationTopicMap.get(notificationPath);
341 public void registerNotificationListenerTopic(SchemaPath notificationPath, TopicId topicId){
342 final Optional<DOMNotificationService> notifyService = netconfMount.getService(DOMNotificationService.class);
343 if(notificationPath != null && notifyService.isPresent()){
344 ListenerRegistration<NetconfEventSource> registration = notifyService.get().registerNotificationListener(this.notificationListener,notificationPath);
345 notificationRegistrationMap.put(notificationPath, registration);
346 ArrayList<TopicId> topicIds = getNotificationTopicIds(notificationPath);
347 if(topicIds == null){
348 topicIds = new ArrayList<>();
349 topicIds.add(topicId);
351 if(topicIds.contains(topicId) == false){
352 topicIds.add(topicId);
355 notificationTopicMap.put(notificationPath, topicIds);