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
8 package org.opendaylight.netconf.messagebus.eventsources.netconf;
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;
36 * Topic registration for notification stream.
38 public class StreamNotificationTopicRegistration extends NotificationTopicRegistration {
40 private static final Logger LOG = LoggerFactory.getLogger(StreamNotificationTopicRegistration.class);
41 private static final NodeIdentifier STREAM_QNAME = NodeIdentifier.create(
42 QName.create(CreateSubscriptionInput.QNAME, "stream"));
43 private static final SchemaPath CREATE_SUBSCRIPTION = SchemaPath
44 .create(true, QName.create(CreateSubscriptionInput.QNAME, "create-subscription"));
45 private static final NodeIdentifier START_TIME_SUBSCRIPTION = NodeIdentifier.create(
46 QName.create(CreateSubscriptionInput.QNAME, "startTime"));
47 private static final NodeIdentifier CREATE_SUBSCRIPTION_INPUT = NodeIdentifier.create(
48 CreateSubscriptionInput.QNAME);
50 final private DOMMountPoint domMountPoint;
51 final private String nodeId;
52 final private NetconfEventSource netconfEventSource;
53 final private Stream stream;
54 private Date lastEventTime;
56 private ConcurrentHashMap<SchemaPath, ListenerRegistration<NetconfEventSource>> notificationRegistrationMap = new ConcurrentHashMap<>();
57 private ConcurrentHashMap<SchemaPath, ArrayList<TopicId>> notificationTopicMap = new ConcurrentHashMap<>();
60 * Creates registration to notification stream.
61 * @param stream stream
62 * @param notificationPrefix notifications namespace
63 * @param netconfEventSource event source
65 public StreamNotificationTopicRegistration(final Stream stream, final String notificationPrefix,
66 NetconfEventSource netconfEventSource) {
67 super(NotificationSourceType.NetconfDeviceStream, stream.getName().getValue(), notificationPrefix);
68 this.domMountPoint = netconfEventSource.getDOMMountPoint();
69 this.nodeId = netconfEventSource.getNode().getNodeId().getValue().toString();
70 this.netconfEventSource = netconfEventSource;
72 this.lastEventTime = null;
73 setReplaySupported(this.stream.isReplaySupport());
75 LOG.info("StreamNotificationTopicRegistration initialized for {}", getStreamName());
79 * Subscribes to notification stream associated with this registration.
81 void activateNotificationSource() {
82 if (isActive() == false) {
83 LOG.info("Stream {} is not active on node {}. Will subscribe.", this.getStreamName(), this.nodeId);
84 final ContainerNode input = Builders.containerBuilder()
85 .withNodeIdentifier(CREATE_SUBSCRIPTION_INPUT)
86 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, this.getStreamName())).build();
87 CheckedFuture<DOMRpcResult, DOMRpcException> csFuture = domMountPoint.getService(DOMRpcService.class).get()
88 .invokeRpc(CREATE_SUBSCRIPTION, input);
90 csFuture.checkedGet();
92 } catch (DOMRpcException e) {
93 LOG.warn("Can not subscribe stream {} on node {}", this.getSourceName(), this.nodeId);
98 LOG.info("Stream {} is now active on node {}", this.getStreamName(), this.nodeId);
103 * Subscribes to notification stream associated with this registration. If replay is supported, notifications from last
104 * received event time will be requested.
106 void reActivateNotificationSource() {
108 LOG.info("Stream {} is reactivating on node {}.", this.getStreamName(), this.nodeId);
109 DataContainerNodeAttrBuilder<NodeIdentifier, ContainerNode> inputBuilder = Builders.containerBuilder()
110 .withNodeIdentifier(CREATE_SUBSCRIPTION_INPUT)
111 .withChild(ImmutableNodes.leafNode(STREAM_QNAME, this.getStreamName()));
112 if (isReplaySupported() && this.getLastEventTime() != null) {
113 inputBuilder.withChild(ImmutableNodes.leafNode(START_TIME_SUBSCRIPTION, this.getLastEventTime()));
115 final ContainerNode input = inputBuilder.build();
116 CheckedFuture<DOMRpcResult, DOMRpcException> csFuture = domMountPoint.getService(DOMRpcService.class).get()
117 .invokeRpc(CREATE_SUBSCRIPTION, input);
119 csFuture.checkedGet();
121 } catch (DOMRpcException e) {
122 LOG.warn("Can not resubscribe stream {} on node {}", this.getSourceName(), this.nodeId);
129 @Override void deActivateNotificationSource() {
130 // no operations need
133 private void closeStream() {
135 for (ListenerRegistration<NetconfEventSource> reg : notificationRegistrationMap.values()) {
138 notificationRegistrationMap.clear();
139 notificationTopicMap.clear();
144 private String getStreamName() {
145 return getSourceName();
148 @Override ArrayList<TopicId> getNotificationTopicIds(SchemaPath notificationPath) {
149 return notificationTopicMap.get(notificationPath);
152 @Override boolean registerNotificationTopic(SchemaPath notificationPath, TopicId topicId) {
154 if (checkNotificationPath(notificationPath) == false) {
155 LOG.debug("Bad SchemaPath for notification try to register");
159 final Optional<DOMNotificationService> notifyService = domMountPoint.getService(DOMNotificationService.class);
160 if (notifyService.isPresent() == false) {
161 LOG.debug("DOMNotificationService is not present");
165 activateNotificationSource();
166 if (isActive() == false) {
167 LOG.warn("Stream {} is not active, listener for notification {} is not registered.", getStreamName(),
168 notificationPath.toString());
172 ListenerRegistration<NetconfEventSource> registration = notifyService.get()
173 .registerNotificationListener(this.netconfEventSource, notificationPath);
174 notificationRegistrationMap.put(notificationPath, registration);
175 ArrayList<TopicId> topicIds = getNotificationTopicIds(notificationPath);
176 if (topicIds == null) {
177 topicIds = new ArrayList<>();
178 topicIds.add(topicId);
180 if (topicIds.contains(topicId) == false) {
181 topicIds.add(topicId);
185 notificationTopicMap.put(notificationPath, topicIds);
189 @Override synchronized void unRegisterNotificationTopic(TopicId topicId) {
190 List<SchemaPath> notificationPathToRemove = new ArrayList<>();
191 for (SchemaPath notifKey : notificationTopicMap.keySet()) {
192 ArrayList<TopicId> topicList = notificationTopicMap.get(notifKey);
193 if (topicList != null) {
194 topicList.remove(topicId);
195 if (topicList.isEmpty()) {
196 notificationPathToRemove.add(notifKey);
200 for (SchemaPath notifKey : notificationPathToRemove) {
201 notificationTopicMap.remove(notifKey);
202 ListenerRegistration<NetconfEventSource> reg = notificationRegistrationMap.remove(notifKey);
209 Optional<Date> getLastEventTime() {
210 return Optional.fromNullable(lastEventTime);
213 void setLastEventTime(Date lastEventTime) {
214 this.lastEventTime = lastEventTime;
217 @Override public void close() throws Exception {