opendaylight/netconf/messagebus-netconf/src/main/java/org/opendaylight/controller/messagebus/eventsources/netconf/StreamNotificationTopicRegistration.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.controller.messagebus.eventsources.netconf;
   9
  10 import com.google.common.base.Optional;
  11 import com.google.common.util.concurrent.CheckedFuture;
  12 import java.util.ArrayList;
  13 import java.util.Date;
  14 import java.util.List;
  15 import java.util.concurrent.ConcurrentHashMap;
  16 import org.opendaylight.controller.md.sal.dom.api.DOMMountPoint;
  17 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
  18 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
  19 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  20 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
  21 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicId;
  22 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.notification._1._0.rev080714.CreateSubscriptionInput;
  23 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netmod.notification.rev080714.netconf.streams.Stream;
  24 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  25 import org.opendaylight.yangtools.yang.common.QName;
  26 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  27 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  28 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  29 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  30 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.DataContainerNodeAttrBuilder;
  31 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 public class StreamNotificationTopicRegistration extends NotificationTopicRegistration {
  36
  37     private static final Logger LOG = LoggerFactory.getLogger(StreamNotificationTopicRegistration.class);
  38     private static final NodeIdentifier STREAM_QNAME = new NodeIdentifier(
  39         QName.create(CreateSubscriptionInput.QNAME, "stream"));
  40     private static final SchemaPath CREATE_SUBSCRIPTION = SchemaPath
  41         .create(true, QName.create(CreateSubscriptionInput.QNAME, "create-subscription"));
  42     private static final NodeIdentifier START_TIME_SUBSCRIPTION = new NodeIdentifier(
  43         QName.create(CreateSubscriptionInput.QNAME, "startTime"));
  44
  45     final private DOMMountPoint domMountPoint;
  46     final private String nodeId;
  47     final private NetconfEventSource netconfEventSource;
  48     final private Stream stream;
  49     private Date lastEventTime;
  50
  51     private ConcurrentHashMap<SchemaPath, ListenerRegistration<NetconfEventSource>> notificationRegistrationMap = new ConcurrentHashMap<>();
  52     private ConcurrentHashMap<SchemaPath, ArrayList<TopicId>> notificationTopicMap = new ConcurrentHashMap<>();
  53
  54     public StreamNotificationTopicRegistration(final Stream stream, final String notificationPrefix,
  55         NetconfEventSource netconfEventSource) {
  56         super(NotificationSourceType.NetconfDeviceStream, stream.getName().getValue(), notificationPrefix);
  57         this.domMountPoint = netconfEventSource.getDOMMountPoint();
  58         this.nodeId = netconfEventSource.getNode().getNodeId().getValue().toString();
  59         this.netconfEventSource = netconfEventSource;
  60         this.stream = stream;
  61         this.lastEventTime = null;
  62         setReplaySupported(this.stream.isReplaySupport());
  63         setActive(false);
  64         LOG.info("StreamNotificationTopicRegistration initialized for {}", getStreamName());
  65     }
  66
  67     void activateNotificationSource() {
  68         if (isActive() == false) {
  69             LOG.info("Stream {} is not active on node {}. Will subscribe.", this.getStreamName(), this.nodeId);
  70             final ContainerNode input = Builders.containerBuilder()
  71                 .withNodeIdentifier(new NodeIdentifier(CreateSubscriptionInput.QNAME))
  72                 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, this.getStreamName())).build();
  73             CheckedFuture<DOMRpcResult, DOMRpcException> csFuture = domMountPoint.getService(DOMRpcService.class).get()
  74                 .invokeRpc(CREATE_SUBSCRIPTION, input);
  75             try {
  76                 csFuture.checkedGet();
  77                 setActive(true);
  78             } catch (DOMRpcException e) {
  79                 LOG.warn("Can not subscribe stream {} on node {}", this.getSourceName(), this.nodeId);
  80                 setActive(false);
  81                 return;
  82             }
  83         } else {
  84             LOG.info("Stream {} is now active on node {}", this.getStreamName(), this.nodeId);
  85         }
  86     }
  87
  88     void reActivateNotificationSource() {
  89         if (isActive()) {
  90             LOG.info("Stream {} is reactivating on node {}.", this.getStreamName(), this.nodeId);
  91             DataContainerNodeAttrBuilder<NodeIdentifier, ContainerNode> inputBuilder = Builders.containerBuilder()
  92                 .withNodeIdentifier(new NodeIdentifier(CreateSubscriptionInput.QNAME))
  93                 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, this.getStreamName()));
  94             if (isReplaySupported() && this.getLastEventTime() != null) {
  95                 inputBuilder.withChild(ImmutableNodes.leafNode(START_TIME_SUBSCRIPTION, this.getLastEventTime()));
  96             }
  97             final ContainerNode input = inputBuilder.build();
  98             CheckedFuture<DOMRpcResult, DOMRpcException> csFuture = domMountPoint.getService(DOMRpcService.class).get()
  99                 .invokeRpc(CREATE_SUBSCRIPTION, input);
 100             try {
 101                 csFuture.checkedGet();
 102                 setActive(true);
 103             } catch (DOMRpcException e) {
 104                 LOG.warn("Can not resubscribe stream {} on node {}", this.getSourceName(), this.nodeId);
 105                 setActive(false);
 106                 return;
 107             }
 108         }
 109     }
 110
 111     @Override void deActivateNotificationSource() {
 112         // no operations need
 113     }
 114
 115     private void closeStream() {
 116         if (isActive()) {
 117             for (ListenerRegistration<NetconfEventSource> reg : notificationRegistrationMap.values()) {
 118                 reg.close();
 119             }
 120             notificationRegistrationMap.clear();
 121             notificationTopicMap.clear();
 122             setActive(false);
 123         }
 124     }
 125
 126     private String getStreamName() {
 127         return getSourceName();
 128     }
 129
 130     @Override ArrayList<TopicId> getNotificationTopicIds(SchemaPath notificationPath) {
 131         return notificationTopicMap.get(notificationPath);
 132     }
 133
 134     @Override boolean registerNotificationTopic(SchemaPath notificationPath, TopicId topicId) {
 135
 136         if (checkNotificationPath(notificationPath) == false) {
 137             LOG.debug("Bad SchemaPath for notification try to register");
 138             return false;
 139         }
 140
 141         final Optional<DOMNotificationService> notifyService = domMountPoint.getService(DOMNotificationService.class);
 142         if (notifyService.isPresent() == false) {
 143             LOG.debug("DOMNotificationService is not present");
 144             return false;
 145         }
 146
 147         activateNotificationSource();
 148         if (isActive() == false) {
 149             LOG.warn("Stream {} is not active, listener for notification {} is not registered.", getStreamName(),
 150                 notificationPath.toString());
 151             return false;
 152         }
 153
 154         ListenerRegistration<NetconfEventSource> registration = notifyService.get()
 155             .registerNotificationListener(this.netconfEventSource, notificationPath);
 156         notificationRegistrationMap.put(notificationPath, registration);
 157         ArrayList<TopicId> topicIds = getNotificationTopicIds(notificationPath);
 158         if (topicIds == null) {
 159             topicIds = new ArrayList<>();
 160             topicIds.add(topicId);
 161         } else {
 162             if (topicIds.contains(topicId) == false) {
 163                 topicIds.add(topicId);
 164             }
 165         }
 166
 167         notificationTopicMap.put(notificationPath, topicIds);
 168         return true;
 169     }
 170
 171     @Override synchronized void unRegisterNotificationTopic(TopicId topicId) {
 172         List<SchemaPath> notificationPathToRemove = new ArrayList<>();
 173         for (SchemaPath notifKey : notificationTopicMap.keySet()) {
 174             ArrayList<TopicId> topicList = notificationTopicMap.get(notifKey);
 175             if (topicList != null) {
 176                 topicList.remove(topicId);
 177                 if (topicList.isEmpty()) {
 178                     notificationPathToRemove.add(notifKey);
 179                 }
 180             }
 181         }
 182         for (SchemaPath notifKey : notificationPathToRemove) {
 183             notificationTopicMap.remove(notifKey);
 184             ListenerRegistration<NetconfEventSource> reg = notificationRegistrationMap.remove(notifKey);
 185             if (reg != null) {
 186                 reg.close();
 187             }
 188         }
 189     }
 190
 191     Optional<Date> getLastEventTime() {
 192         return Optional.fromNullable(lastEventTime);
 193     }
 194
 195     void setLastEventTime(Date lastEventTime) {
 196         this.lastEventTime = lastEventTime;
 197     }
 198
 199     @Override public void close() throws Exception {
 200         closeStream();
 201     }
 202
 203 }