--- /dev/null
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.messagebus.app.impl;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Future;
+import java.util.regex.Pattern;
+import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
+import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
+import org.opendaylight.controller.mdsal.MdSAL;
+import org.opendaylight.controller.sal.core.api.notify.NotificationListener;
+import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.NotificationPattern;
+import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicNotification;
+import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.EventSourceService;
+import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicInput;
+import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutput;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInputBuilder;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.NotificationsService;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.StreamNameType;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.inventory.rev140108.NetconfNode;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.binding.DataObject;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode;
+import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NetconfEventSource implements EventSourceService, NotificationListener, DataChangeListener {
+ private static final Logger LOGGER = LoggerFactory.getLogger(NetconfEventSource.class);
+
+ private final MdSAL mdSal;
+ private final String nodeId;
+
+ private final List<String> activeStreams = new ArrayList<>();
+
+ private final Map<String, String> urnPrefixToStreamMap;
+
+ public NetconfEventSource(final MdSAL mdSal, final String nodeId, final Map<String, String> streamMap) {
+ Preconditions.checkNotNull(mdSal);
+ Preconditions.checkNotNull(nodeId);
+
+ this.mdSal = mdSal;
+ this.nodeId = nodeId;
+ this.urnPrefixToStreamMap = streamMap;
+
+ LOGGER.info("NetconfEventSource [{}] created.", nodeId);
+ }
+
+ @Override
+ public Future<RpcResult<JoinTopicOutput>> joinTopic(final JoinTopicInput input) {
+ final NotificationPattern notificationPattern = input.getNotificationPattern();
+
+ // FIXME: default language should already be regex
+ final String regex = Util.wildcardToRegex(notificationPattern.getValue());
+
+ final Pattern pattern = Pattern.compile(regex);
+ List<QName> matchingNotifications = Util.expandQname(availableNotifications(), pattern);
+ registerNotificationListener(matchingNotifications);
+ return null;
+ }
+
+ private List<QName> availableNotifications() {
+ // FIXME: use SchemaContextListener to get changes asynchronously
+ Set<NotificationDefinition> availableNotifications = mdSal.getSchemaContext(nodeId).getNotifications();
+ List<QName> qNs = new ArrayList<>(availableNotifications.size());
+ for (NotificationDefinition nd : availableNotifications) {
+ qNs.add(nd.getQName());
+ }
+
+ return qNs;
+ }
+
+ private void registerNotificationListener(final List<QName> notificationsToSubscribe) {
+ for (QName qName : notificationsToSubscribe) {
+ startSubscription(qName);
+ // FIXME: do not lose this registration
+ final ListenerRegistration<NotificationListener> reg = mdSal.addNotificationListener(nodeId, qName, this);
+ }
+ }
+
+ private synchronized void startSubscription(final QName qName) {
+ String streamName = resolveStream(qName);
+
+ if (streamIsActive(streamName) == false) {
+ LOGGER.info("Stream {} is not active on node {}. Will subscribe.", streamName, nodeId);
+ startSubscription(streamName);
+ }
+ }
+
+ private synchronized void resubscribeToActiveStreams() {
+ for (String streamName : activeStreams) {
+ startSubscription(streamName);
+ }
+ }
+
+ private synchronized void startSubscription(final String streamName) {
+ CreateSubscriptionInput subscriptionInput = getSubscriptionInput(streamName);
+ mdSal.getRpcService(nodeId, NotificationsService.class).createSubscription(subscriptionInput);
+ activeStreams.add(streamName);
+ }
+
+ private static CreateSubscriptionInput getSubscriptionInput(final String streamName) {
+ CreateSubscriptionInputBuilder csib = new CreateSubscriptionInputBuilder();
+ csib.setStream(new StreamNameType(streamName));
+ return csib.build();
+ }
+
+ private String resolveStream(final QName qName) {
+ String streamName = null;
+
+ for (Map.Entry<String, String> entry : urnPrefixToStreamMap.entrySet()) {
+ String nameSpace = qName.getNamespace().toString();
+ String urnPrefix = entry.getKey();
+ if( nameSpace.startsWith(urnPrefix) ) {
+ streamName = entry.getValue();
+ break;
+ }
+ }
+
+ return streamName;
+ }
+
+ private boolean streamIsActive(final String streamName) {
+ return activeStreams.contains(streamName);
+ }
+
+ // PASS
+ @Override public Set<QName> getSupportedNotifications() {
+ return null;
+ }
+
+ @Override
+ public void onNotification(final CompositeNode notification) {
+ LOGGER.info("NetconfEventSource {} received notification {}. Will publish to MD-SAL.", nodeId, notification);
+ ImmutableCompositeNode payload = ImmutableCompositeNode.builder()
+ .setQName(QName.create(TopicNotification.QNAME, "payload"))
+ .add(notification).toInstance();
+ ImmutableCompositeNode icn = ImmutableCompositeNode.builder()
+ .setQName(TopicNotification.QNAME)
+ .add(payload)
+ .addLeaf("event-source", nodeId)
+ .toInstance();
+
+ mdSal.publishNotification(icn);
+ }
+
+ @Override
+ public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> change) {
+ boolean wasConnected = false;
+ boolean nowConnected = false;
+
+ for (Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getOriginalData().entrySet()) {
+ if ( isNetconfNode(changeEntry) ) {
+ NetconfNode nn = (NetconfNode)changeEntry.getValue();
+ wasConnected = nn.isConnected();
+ }
+ }
+
+ for (Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : change.getUpdatedData().entrySet()) {
+ if ( isNetconfNode(changeEntry) ) {
+ NetconfNode nn = (NetconfNode)changeEntry.getValue();
+ nowConnected = nn.isConnected();
+ }
+ }
+
+ if (wasConnected == false && nowConnected == true) {
+ resubscribeToActiveStreams();
+ }
+ }
+
+ private static boolean isNetconfNode(final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry ) {
+ return NetconfNode.class.equals(changeEntry.getKey().getTargetType());
+ }
+
+}
Code Review / controller.git / blobdiff