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.app.impl;
12 import java.util.regex.Pattern;
14 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
15 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
16 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.NotificationPattern;
17 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.TopicId;
18 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.EventSourceService;
19 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicInput;
20 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicInputBuilder;
21 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.JoinTopicOutput;
22 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
23 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
24 import org.opendaylight.yangtools.yang.binding.DataObject;
25 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
26 import org.opendaylight.yangtools.yang.common.RpcError;
27 import org.opendaylight.yangtools.yang.common.RpcResult;
28 import org.slf4j.LoggerFactory;
30 import com.google.common.base.Preconditions;
32 public class EventSourceTopic implements DataChangeListener {
33 private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(EventSourceTopic.class);
34 private final NotificationPattern notificationPattern;
35 private final EventSourceService sourceService;
36 private final Pattern nodeIdPattern;
37 private final TopicId topicId;
39 public EventSourceTopic(final NotificationPattern notificationPattern, final String nodeIdPattern, final EventSourceService eventSource) {
40 this.notificationPattern = Preconditions.checkNotNull(notificationPattern);
41 this.sourceService = eventSource;
43 // FIXME: regex should be the language of nodeIdPattern
44 final String regex = Util.wildcardToRegex(nodeIdPattern);
45 this.nodeIdPattern = Pattern.compile(regex);
47 this.topicId = new TopicId(Util.getUUIDIdent());
50 public TopicId getTopicId() {
55 public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> event) {
56 for (final Map.Entry<InstanceIdentifier<?>, DataObject> changeEntry : event.getUpdatedData().entrySet()) {
57 if (changeEntry.getValue() instanceof Node) {
58 final Node node = (Node) changeEntry.getValue();
59 if (nodeIdPattern.matcher(node.getId().getValue()).matches()) {
60 notifyNode(changeEntry.getKey());
66 public void notifyNode(final InstanceIdentifier<?> nodeId) {
69 RpcResult<JoinTopicOutput> rpcResultJoinTopic = sourceService.joinTopic(getJoinTopicInputArgument(nodeId)).get();
70 if(rpcResultJoinTopic.isSuccessful() == false){
71 for(RpcError err : rpcResultJoinTopic.getErrors()){
72 LOG.error("Can not join topic: [{}] on node: [{}]. Error: {}",getTopicId().getValue(),nodeId.toString(),err.toString());
75 } catch (final Exception e) {
76 LOG.error("Could not invoke join topic for node {}", nodeId);
80 private JoinTopicInput getJoinTopicInputArgument(final InstanceIdentifier<?> path) {
81 final NodeRef nodeRef = new NodeRef(path);
82 final JoinTopicInput jti =
83 new JoinTopicInputBuilder()
84 .setNode(nodeRef.getValue())
86 .setNotificationPattern(notificationPattern)