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.ConcurrentSkipListSet;
20 import java.util.concurrent.Future;
21 import java.util.regex.Pattern;
23 import javax.xml.stream.XMLStreamException;
24 import javax.xml.transform.dom.DOMResult;
25 import javax.xml.transform.dom.DOMSource;
27 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
28 import org.opendaylight.controller.md.sal.binding.api.MountPoint;
29 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
30 import org.opendaylight.controller.md.sal.dom.api.DOMMountPoint;
31 import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
32 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
33 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
34 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
35 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
36 import org.opendaylight.controller.messagebus.app.impl.TopicDOMNotification;
37 import org.opendaylight.controller.messagebus.app.impl.Util;
38 import org.opendaylight.controller.messagebus.spi.EventSource;
39 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
40 import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
41 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.NotificationPattern;
42 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicId;
43 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicNotification;
44 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicInput;
45 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutput;
46 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutputBuilder;
47 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicStatus;
48 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.inventory.rev140108.NetconfNode;
50 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
51 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
52 import org.opendaylight.yangtools.concepts.ListenerRegistration;
53 import org.opendaylight.yangtools.yang.binding.DataObject;
54 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
55 import org.opendaylight.yangtools.yang.common.QName;
56 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
57 import org.opendaylight.yangtools.yang.common.RpcResult;
58 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
59 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
60 import org.opendaylight.yangtools.yang.data.api.schema.AnyXmlNode;
61 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
62 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
63 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
64 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
65 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
66 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
67 import org.slf4j.Logger;
68 import org.slf4j.LoggerFactory;
69 import org.w3c.dom.Document;
70 import org.w3c.dom.Element;
72 import com.google.common.base.Optional;
73 import com.google.common.base.Throwables;
75 public class NetconfEventSource implements EventSource, DOMNotificationListener, DataChangeListener {
77 private static final Logger LOG = LoggerFactory.getLogger(NetconfEventSource.class);
79 private static final NodeIdentifier TOPIC_NOTIFICATION_ARG = new NodeIdentifier(TopicNotification.QNAME);
80 private static final NodeIdentifier EVENT_SOURCE_ARG = new NodeIdentifier(QName.create(TopicNotification.QNAME, "node-id"));
81 private static final NodeIdentifier PAYLOAD_ARG = new NodeIdentifier(QName.create(TopicNotification.QNAME, "payload"));
83 private static final NodeIdentifier STREAM_QNAME = new NodeIdentifier(QName.create(CreateSubscriptionInput.QNAME,"stream"));
84 private static final SchemaPath CREATE_SUBSCRIPTION = SchemaPath.create(true, QName.create(CreateSubscriptionInput.QNAME, "create-subscription"));
86 private final String nodeId;
87 private final Node node;
89 private final DOMMountPoint netconfMount;
90 private final DOMNotificationPublishService domPublish;
91 private final Set<String> activeStreams = new ConcurrentSkipListSet<>();
93 private final Map<String, String> urnPrefixToStreamMap;
94 private final ConcurrentHashMap<TopicId,ListenerRegistration<NetconfEventSource>> listenerRegistrationMap = new ConcurrentHashMap<>();
96 public NetconfEventSource(final Node node, final Map<String, String> streamMap, final DOMMountPoint netconfMount, final DOMNotificationPublishService publishService, final MountPoint bindingMount) {
97 this.netconfMount = netconfMount;
99 this.nodeId = node.getNodeId().getValue();
100 this.urnPrefixToStreamMap = streamMap;
101 this.domPublish = publishService;
102 LOG.info("NetconfEventSource [{}] created.", nodeId);
106 public Future<RpcResult<JoinTopicOutput>> joinTopic(final JoinTopicInput input) {
107 final NotificationPattern notificationPattern = input.getNotificationPattern();
108 final List<SchemaPath> matchingNotifications = getMatchingNotifications(notificationPattern);
109 return registerNotificationListener(input.getTopicId(),matchingNotifications);
112 private synchronized Future<RpcResult<JoinTopicOutput>> registerNotificationListener(final TopicId topicId, final List<SchemaPath> notificationsToSubscribe){
113 if(listenerRegistrationMap.containsKey(topicId)){
114 final String errMessage = "Can not join topic twice. Topic " + topicId.getValue() + " has been joined to node " + this.nodeId;
115 return immediateFuture(RpcResultBuilder.<JoinTopicOutput>failed().withError(ErrorType.APPLICATION, errMessage).build());
117 ListenerRegistration<NetconfEventSource> registration = null;
118 JoinTopicStatus joinTopicStatus = JoinTopicStatus.Down;
119 final Optional<DOMNotificationService> notifyService = netconfMount.getService(DOMNotificationService.class);
121 if(notifyService.isPresent()) {
122 for (final SchemaPath qName : notificationsToSubscribe) {
123 startSubscription(qName);
125 registration = notifyService.get().registerNotificationListener(this, notificationsToSubscribe);
128 if(registration != null){
129 listenerRegistrationMap.put(topicId,registration);
130 joinTopicStatus = JoinTopicStatus.Up;
132 final JoinTopicOutput output = new JoinTopicOutputBuilder().setStatus(joinTopicStatus).build();
133 return immediateFuture(RpcResultBuilder.success(output).build());
136 private void startSubscription(final SchemaPath path) {
137 final String streamName = resolveStream(path.getLastComponent());
138 startSubscription(streamName);
141 private void resubscribeToActiveStreams() {
142 for (final String streamName : activeStreams) {
143 startSubscription(streamName);
147 private synchronized void startSubscription(final String streamName) {
148 if(streamIsActive(streamName) == false){
149 LOG.info("Stream {} is not active on node {}. Will subscribe.", streamName, nodeId);
150 final ContainerNode input = Builders.containerBuilder().withNodeIdentifier(new NodeIdentifier(CreateSubscriptionInput.QNAME))
151 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, streamName))
153 netconfMount.getService(DOMRpcService.class).get().invokeRpc(CREATE_SUBSCRIPTION, input);
154 activeStreams.add(streamName);
158 private String resolveStream(final QName qName) {
159 String streamName = null;
161 for (final Map.Entry<String, String> entry : urnPrefixToStreamMap.entrySet()) {
162 final String nameSpace = qName.getNamespace().toString();
163 final String urnPrefix = entry.getKey();
164 if( nameSpace.startsWith(urnPrefix) ) {
165 streamName = entry.getValue();
173 private boolean streamIsActive(final String streamName) {
174 return activeStreams.contains(streamName);
178 public void onNotification(final DOMNotification notification) {
179 final ContainerNode topicNotification = Builders.containerBuilder()
180 .withNodeIdentifier(TOPIC_NOTIFICATION_ARG)
181 .withChild(ImmutableNodes.leafNode(EVENT_SOURCE_ARG, nodeId))
182 .withChild(encapsulate(notification))
185 domPublish.putNotification(new TopicDOMNotification(topicNotification));
186 } catch (final InterruptedException e) {
187 throw Throwables.propagate(e);
191 private AnyXmlNode encapsulate(final DOMNotification body) {
192 // FIXME: Introduce something like AnyXmlWithNormalizedNodeData in Yangtools
193 final Document doc = XmlUtil.newDocument();
194 final Optional<String> namespace = Optional.of(PAYLOAD_ARG.getNodeType().getNamespace().toString());
195 final Element element = XmlUtil.createElement(doc , "payload", namespace);
197 final DOMResult result = new DOMResult(element);
199 final SchemaContext context = netconfMount.getSchemaContext();
200 final SchemaPath schemaPath = body.getType();
202 NetconfMessageTransformUtil.writeNormalizedNode(body.getBody(), result, schemaPath, context);
203 return Builders.anyXmlBuilder().withNodeIdentifier(PAYLOAD_ARG)
204 .withValue(new DOMSource(element))
206 } catch (IOException | XMLStreamException e) {
207 LOG.error("Unable to encapsulate notification.",e);
208 throw Throwables.propagate(e);
213 public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> change) {
214 boolean wasConnected = false;
215 boolean nowConnected = false;
217 for (final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getOriginalData().entrySet()) {
218 if ( isNetconfNode(changeEntry) ) {
219 final NetconfNode nn = (NetconfNode)changeEntry.getValue();
220 wasConnected = nn.isConnected();
224 for (final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getUpdatedData().entrySet()) {
225 if ( isNetconfNode(changeEntry) ) {
226 final NetconfNode nn = (NetconfNode)changeEntry.getValue();
227 nowConnected = nn.isConnected();
231 if (wasConnected == false && nowConnected == true) {
232 resubscribeToActiveStreams();
236 private static boolean isNetconfNode(final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry ) {
237 return NetconfNode.class.equals(changeEntry.getKey().getTargetType());
240 private List<SchemaPath> getMatchingNotifications(NotificationPattern notificationPattern){
241 // FIXME: default language should already be regex
242 final String regex = Util.wildcardToRegex(notificationPattern.getValue());
244 final Pattern pattern = Pattern.compile(regex);
245 return Util.expandQname(getAvailableNotifications(), pattern);
249 public void close() throws Exception {
250 for(ListenerRegistration<NetconfEventSource> registration : listenerRegistrationMap.values()){
251 registration.close();
256 public NodeKey getSourceNodeKey(){
257 return node.getKey();
261 public List<SchemaPath> getAvailableNotifications() {
262 // FIXME: use SchemaContextListener to get changes asynchronously
263 final Set<NotificationDefinition> availableNotifications = netconfMount.getSchemaContext().getNotifications();
264 final List<SchemaPath> qNs = new ArrayList<>(availableNotifications.size());
265 for (final NotificationDefinition nd : availableNotifications) {
266 qNs.add(nd.getPath());