2 * Copyright (c) 2014 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.openflowplugin.applications.statistics.manager.impl;
11 import com.google.common.base.Optional;
12 import com.google.common.base.Preconditions;
14 import java.util.concurrent.ConcurrentHashMap;
15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
16 import org.opendaylight.controller.md.sal.binding.api.DataChangeListener;
17 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
18 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
19 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
20 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
21 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
22 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
23 import org.opendaylight.openflowplugin.applications.statistics.manager.StatListeningCommiter;
24 import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
25 import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
28 import org.opendaylight.yangtools.concepts.ListenerRegistration;
29 import org.opendaylight.yangtools.yang.binding.DataObject;
30 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
31 import org.opendaylight.yangtools.yang.binding.NotificationListener;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
37 * org.opendaylight.openflowplugin.applications.statistics.manager.impl
39 * StatAbstractListeneningCommiter
40 * Class is abstract implementation for all Configuration/DataStore DataChange
41 * listenable DataObjects like flows, groups, meters. It is a holder for common
42 * functionality needed by construction/destruction class and for DataChange
46 public abstract class StatAbstractListenCommit<T extends DataObject, N extends NotificationListener>
47 extends StatAbstractNotifyCommit<N> implements StatListeningCommiter<T,N> {
49 private static final Logger LOG = LoggerFactory.getLogger(StatAbstractListenCommit.class);
51 private ListenerRegistration<DataChangeListener> listenerRegistration;
53 protected final Map<InstanceIdentifier<Node>, Map<InstanceIdentifier<T>, Integer>> mapNodesForDelete = new ConcurrentHashMap<>();
54 protected final Map<InstanceIdentifier<Node>, Integer> mapNodeFeautureRepeater = new ConcurrentHashMap<>();
56 private final Class<T> clazz;
58 private final DataBroker dataBroker;
60 protected final StatNodeRegistration nodeRegistrationManager;
62 private ReadOnlyTransaction currentReadTx;
63 private volatile boolean currentReadTxStale;
65 /* Constructor has to make a registration */
66 public StatAbstractListenCommit(final StatisticsManager manager, final DataBroker db,
67 final NotificationProviderService nps, final Class<T> clazz, final StatNodeRegistration nodeRegistrationManager) {
68 super(manager,nps, nodeRegistrationManager);
69 this.clazz = Preconditions.checkNotNull(clazz, "Referenced Class can not be null");
70 Preconditions.checkArgument(db != null, "DataBroker can not be null!");
71 listenerRegistration = db.registerDataChangeListener(LogicalDatastoreType.CONFIGURATION,
72 getWildCardedRegistrationPath(), this, DataChangeScope.BASE);
74 this.nodeRegistrationManager = nodeRegistrationManager;
78 * Method returns WildCarded Path which is used for registration as a listening path changes in
79 * {@link org.opendaylight.controller.md.sal.binding.api.DataChangeListener}
82 protected abstract InstanceIdentifier<T> getWildCardedRegistrationPath();
85 public void onDataChanged(final AsyncDataChangeEvent<InstanceIdentifier<?>, DataObject> changeEvent) {
86 Preconditions.checkNotNull(changeEvent,"Async ChangeEvent can not be null!");
89 * If we have opened read transaction for configuration data store, we need to mark it as stale.
91 * Latest read transaction will be allocated on another read using readLatestConfiguration
93 currentReadTxStale = true;
96 @SuppressWarnings("unchecked")
97 protected void removeData(final InstanceIdentifier<?> key, final Integer value) {
98 if (clazz.equals(key.getTargetType())) {
99 final InstanceIdentifier<Node> nodeIdent = key.firstIdentifierOf(Node.class);
100 Map<InstanceIdentifier<T>, Integer> map = null;
101 if (mapNodesForDelete.containsKey(nodeIdent)) {
102 map = mapNodesForDelete.get(nodeIdent);
105 map = new ConcurrentHashMap<>();
106 mapNodesForDelete.put(nodeIdent, map);
108 map.put((InstanceIdentifier<T>) key, value);
113 public void cleanForDisconnect(final InstanceIdentifier<Node> nodeIdent) {
114 mapNodesForDelete.remove(nodeIdent);
118 public void close() {
119 if (listenerRegistration != null) {
121 listenerRegistration.close();
122 } catch (final Exception e) {
123 LOG.error("Error by stop {} DataChange StatListeningCommiter.", clazz.getSimpleName(), e);
125 listenerRegistration = null;
132 * Method return actual DataObject identified by InstanceIdentifier from Config/DS
136 protected final <K extends DataObject> Optional<K> readLatestConfiguration(final InstanceIdentifier<K> path) {
137 for(int i = 0; i < 2; i++) {
138 boolean localReadTxStale = currentReadTxStale;
140 // This non-volatile read piggy backs the volatile currentReadTxStale read above to
141 // ensure visibility in case this method is called across threads (although not concurrently).
142 ReadOnlyTransaction localReadTx = currentReadTx;
143 if(localReadTx == null || localReadTxStale) {
144 if(localReadTx != null) {
148 localReadTx = dataBroker.newReadOnlyTransaction();
150 currentReadTx = localReadTx;
152 // Note - this volatile write also publishes the non-volatile currentReadTx write above.
153 currentReadTxStale = false;
157 return localReadTx.read(LogicalDatastoreType.CONFIGURATION, path).checkedGet();
158 } catch (final ReadFailedException e) {
159 LOG.debug("It wasn't possible to read {} from datastore. Exception: {}", path, e);
161 // Loop back and try again with a new Tx.
162 currentReadTxStale = true;
166 return Optional.absent();