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;
11 import java.util.List;
13 import java.util.concurrent.ConcurrentHashMap;
14 import java.util.concurrent.Future;
15 import java.util.regex.Pattern;
17 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
18 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
19 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
20 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
21 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
22 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
23 import org.opendaylight.controller.messagebus.spi.EventSource;
24 import org.opendaylight.controller.messagebus.spi.EventSourceRegistration;
25 import org.opendaylight.controller.messagebus.spi.EventSourceRegistry;
26 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
27 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
28 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
29 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.CreateTopicInput;
30 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.CreateTopicOutput;
31 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.CreateTopicOutputBuilder;
32 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.DestroyTopicInput;
33 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.EventAggregatorService;
34 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventaggregator.rev141202.NotificationPattern;
35 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.EventSourceService;
36 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.Node1;
37 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.Node1Builder;
38 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.TopologyTypes1;
39 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.TopologyTypes1Builder;
40 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.topology.event.source.type.TopologyEventSource;
41 import org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.eventsource.rev141202.topology.event.source.type.TopologyEventSourceBuilder;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeContext;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
44 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
45 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
46 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
47 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
48 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
49 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
50 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypes;
51 import org.opendaylight.yangtools.concepts.ListenerRegistration;
52 import org.opendaylight.yangtools.yang.binding.DataObject;
53 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
54 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
55 import org.opendaylight.yangtools.yang.common.RpcResult;
56 import org.slf4j.Logger;
57 import org.slf4j.LoggerFactory;
59 import com.google.common.base.Optional;
60 import com.google.common.util.concurrent.CheckedFuture;
61 import com.google.common.util.concurrent.FutureCallback;
62 import com.google.common.util.concurrent.Futures;
65 public class EventSourceTopology implements EventAggregatorService, EventSourceRegistry {
66 private static final Logger LOG = LoggerFactory.getLogger(EventSourceTopology.class);
68 private static final String TOPOLOGY_ID = "EVENT-SOURCE-TOPOLOGY" ;
69 private static final TopologyKey EVENT_SOURCE_TOPOLOGY_KEY = new TopologyKey(new TopologyId(TOPOLOGY_ID));
70 private static final LogicalDatastoreType OPERATIONAL = LogicalDatastoreType.OPERATIONAL;
72 private static final InstanceIdentifier<Topology> EVENT_SOURCE_TOPOLOGY_PATH =
73 InstanceIdentifier.create(NetworkTopology.class)
74 .child(Topology.class, EVENT_SOURCE_TOPOLOGY_KEY);
76 private static final InstanceIdentifier<TopologyTypes1> TOPOLOGY_TYPE_PATH =
77 EVENT_SOURCE_TOPOLOGY_PATH
78 .child(TopologyTypes.class)
79 .augmentation(TopologyTypes1.class);
81 private final Map<DataChangeListener, ListenerRegistration<DataChangeListener>> topicListenerRegistrations =
82 new ConcurrentHashMap<>();
83 private final Map<NodeKey, RoutedRpcRegistration<EventSourceService>> routedRpcRegistrations =
84 new ConcurrentHashMap<>();;
86 private final DataBroker dataBroker;
87 private final RpcRegistration<EventAggregatorService> aggregatorRpcReg;
88 private final EventSourceService eventSourceService;
89 private final RpcProviderRegistry rpcRegistry;
91 public EventSourceTopology(final DataBroker dataBroker, final RpcProviderRegistry rpcRegistry) {
93 this.dataBroker = dataBroker;
94 this.rpcRegistry = rpcRegistry;
95 aggregatorRpcReg = rpcRegistry.addRpcImplementation(EventAggregatorService.class, this);
96 eventSourceService = rpcRegistry.getRpcService(EventSourceService.class);
98 final TopologyEventSource topologySource = new TopologyEventSourceBuilder().build();
99 final TopologyTypes1 topologyTypeAugment = new TopologyTypes1Builder().setTopologyEventSource(topologySource).build();
100 putData(OPERATIONAL, TOPOLOGY_TYPE_PATH, topologyTypeAugment);
101 LOG.info("EventSourceRegistry has been initialized");
104 private <T extends DataObject> void putData(final LogicalDatastoreType store,
105 final InstanceIdentifier<T> path,
108 final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
109 tx.put(store, path, data, true);
114 private <T extends DataObject> void deleteData(final LogicalDatastoreType store, final InstanceIdentifier<T> path){
115 final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
116 tx.delete(OPERATIONAL, path);
120 private void insert(final KeyedInstanceIdentifier<Node, NodeKey> sourcePath) {
121 final NodeKey nodeKey = sourcePath.getKey();
122 final InstanceIdentifier<Node1> augmentPath = sourcePath.augmentation(Node1.class);
123 final Node1 nodeAgument = new Node1Builder().setEventSourceNode(new NodeId(nodeKey.getNodeId().getValue())).build();
124 putData(OPERATIONAL, augmentPath, nodeAgument);
127 private void remove(final KeyedInstanceIdentifier<Node, NodeKey> sourcePath){
128 final InstanceIdentifier<Node1> augmentPath = sourcePath.augmentation(Node1.class);
129 deleteData(OPERATIONAL, augmentPath);
132 private void notifyExistingNodes(final Pattern nodeIdPatternRegex, final EventSourceTopic eventSourceTopic){
134 final ReadOnlyTransaction tx = dataBroker.newReadOnlyTransaction();
136 final CheckedFuture<Optional<Topology>, ReadFailedException> future = tx.read(OPERATIONAL, EVENT_SOURCE_TOPOLOGY_PATH);
138 Futures.addCallback(future, new FutureCallback<Optional<Topology>>(){
141 public void onSuccess(Optional<Topology> data) {
142 if(data.isPresent()) {
143 final List<Node> nodes = data.get().getNode();
144 for (final Node node : nodes) {
145 if (nodeIdPatternRegex.matcher(node.getNodeId().getValue()).matches()) {
146 eventSourceTopic.notifyNode(EVENT_SOURCE_TOPOLOGY_PATH.child(Node.class, node.getKey()));
154 public void onFailure(Throwable t) {
155 LOG.error("Can not notify existing nodes {}", t);
164 public Future<RpcResult<CreateTopicOutput>> createTopic(final CreateTopicInput input) {
165 LOG.info("Received Topic creation request: NotificationPattern -> {}, NodeIdPattern -> {}",
166 input.getNotificationPattern(),
167 input.getNodeIdPattern());
169 final NotificationPattern notificationPattern = new NotificationPattern(input.getNotificationPattern());
170 final String nodeIdPattern = input.getNodeIdPattern().getValue();
171 final Pattern nodeIdPatternRegex = Pattern.compile(Util.wildcardToRegex(nodeIdPattern));
172 final EventSourceTopic eventSourceTopic = new EventSourceTopic(notificationPattern, nodeIdPattern, eventSourceService);
174 registerTopic(eventSourceTopic);
176 notifyExistingNodes(nodeIdPatternRegex, eventSourceTopic);
178 final CreateTopicOutput cto = new CreateTopicOutputBuilder()
179 .setTopicId(eventSourceTopic.getTopicId())
182 return Util.resultRpcSuccessFor(cto);
186 public Future<RpcResult<Void>> destroyTopic(final DestroyTopicInput input) {
187 return Futures.immediateFailedFuture(new UnsupportedOperationException("Not Implemented"));
191 public void close() {
192 aggregatorRpcReg.close();
193 for(ListenerRegistration<DataChangeListener> reg : topicListenerRegistrations.values()){
198 private void registerTopic(final EventSourceTopic listener) {
199 final ListenerRegistration<DataChangeListener> listenerRegistration = dataBroker.registerDataChangeListener(OPERATIONAL,
200 EVENT_SOURCE_TOPOLOGY_PATH,
202 DataBroker.DataChangeScope.SUBTREE);
204 topicListenerRegistrations.put(listener, listenerRegistration);
207 public void register(final EventSource eventSource){
208 NodeKey nodeKey = eventSource.getSourceNodeKey();
209 final KeyedInstanceIdentifier<Node, NodeKey> sourcePath = EVENT_SOURCE_TOPOLOGY_PATH.child(Node.class, nodeKey);
210 RoutedRpcRegistration<EventSourceService> reg = rpcRegistry.addRoutedRpcImplementation(EventSourceService.class, eventSource);
211 reg.registerPath(NodeContext.class, sourcePath);
212 routedRpcRegistrations.put(nodeKey,reg);
216 public void unRegister(final EventSource eventSource){
217 final NodeKey nodeKey = eventSource.getSourceNodeKey();
218 final KeyedInstanceIdentifier<Node, NodeKey> sourcePath = EVENT_SOURCE_TOPOLOGY_PATH.child(Node.class, nodeKey);
219 final RoutedRpcRegistration<EventSourceService> removeRegistration = routedRpcRegistrations.remove(nodeKey);
220 if(removeRegistration != null){
221 removeRegistration.close();
227 public <T extends EventSource> EventSourceRegistration<T> registerEventSource(
229 EventSourceRegistrationImpl<T> esr = new EventSourceRegistrationImpl<>(eventSource, this);
230 register(eventSource);