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;
13 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
14 import org.opendaylight.controller.md.sal.binding.api.DataTreeIdentifier;
15 import org.opendaylight.controller.md.sal.binding.api.DataTreeModification;
16 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
17 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
18 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
19 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
20 import org.opendaylight.openflowplugin.applications.statistics.manager.StatListeningCommiter;
21 import org.opendaylight.openflowplugin.applications.statistics.manager.StatNodeRegistration;
22 import org.opendaylight.openflowplugin.applications.statistics.manager.StatisticsManager;
23 import org.opendaylight.openflowplugin.common.wait.SimpleTaskRetryLooper;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
25 import org.opendaylight.yangtools.concepts.ListenerRegistration;
26 import org.opendaylight.yangtools.yang.binding.DataObject;
27 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
28 import org.opendaylight.yangtools.yang.binding.NotificationListener;
29 import org.slf4j.Logger;
30 import org.slf4j.LoggerFactory;
32 import java.util.ArrayList;
33 import java.util.Collection;
35 import java.util.concurrent.Callable;
36 import java.util.concurrent.ConcurrentHashMap;
40 * org.opendaylight.openflowplugin.applications.statistics.manager.impl
42 * StatAbstractListeneningCommiter
43 * Class is abstract implementation for all Configuration/DataStore DataTreeModification
44 * listenable DataObjects like flows, groups, meters. It is a holder for common
45 * functionality needed by construction/destruction class and for DataTreeModification
48 public abstract class StatAbstractListenCommit<T extends DataObject, N extends NotificationListener>
49 extends StatAbstractNotifyCommit<N> implements StatListeningCommiter<T,N> {
51 private static final Logger LOG = LoggerFactory.getLogger(StatAbstractListenCommit.class);
53 private ListenerRegistration<StatAbstractListenCommit<T, N>> listenerRegistration;
55 protected final Map<InstanceIdentifier<Node>, Map<InstanceIdentifier<T>, Integer>> mapNodesForDelete = new ConcurrentHashMap<>();
56 protected final Map<InstanceIdentifier<Node>, Integer> mapNodeFeautureRepeater = new ConcurrentHashMap<>();
57 protected final Map<InstanceIdentifier<Node>, ArrayList<T>> removedDataBetweenStatsCycle = new
58 ConcurrentHashMap<>();
60 private final Class<T> clazz;
62 protected final DataBroker dataBroker;
64 protected final StatNodeRegistration nodeRegistrationManager;
66 private ReadOnlyTransaction currentReadTx;
67 private volatile boolean currentReadTxStale;
69 private static final int STARTUP_LOOP_TICK = 500;
70 private static final int STARTUP_LOOP_MAX_RETRIES = 8;
72 private final DataTreeIdentifier<T> treeId =
73 new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION, getWildCardedRegistrationPath());
75 /* Constructor has to make a registration */
76 public StatAbstractListenCommit(final StatisticsManager manager, final DataBroker db,
77 final NotificationProviderService nps, final Class<T> clazz, final StatNodeRegistration nodeRegistrationManager) {
78 super(manager,nps, nodeRegistrationManager);
79 this.clazz = Preconditions.checkNotNull(clazz, "Referenced Class can not be null");
80 Preconditions.checkArgument(db != null, "DataBroker can not be null!");
82 this.nodeRegistrationManager = nodeRegistrationManager;
84 SimpleTaskRetryLooper looper = new SimpleTaskRetryLooper(STARTUP_LOOP_TICK, STARTUP_LOOP_MAX_RETRIES);
86 listenerRegistration = looper.loopUntilNoException(new Callable<ListenerRegistration<StatAbstractListenCommit<T, N>>>() {
88 public ListenerRegistration<StatAbstractListenCommit<T, N>> call() throws Exception {
89 return db.registerDataTreeChangeListener(treeId,StatAbstractListenCommit.this);
92 } catch (final Exception ex) {
93 LOG.debug(" StatAbstractListenCommit DataTreeChangeListener registration failed {}", ex.getMessage());
94 throw new IllegalStateException("Notification supplier startup fail! System needs restart.", ex);
99 * Method returns WildCarded Path which is used for registration as a listening path changes in
100 * {@link org.opendaylight.controller.md.sal.binding.api.DataTreeChangeListener}
103 protected abstract InstanceIdentifier<T> getWildCardedRegistrationPath();
105 protected abstract void processDataChange(Collection<DataTreeModification<T>> changes);
108 public void onDataTreeChanged(Collection<DataTreeModification<T>> changes) {
109 Preconditions.checkNotNull(changes, "Changes must not be null!");
111 * If we have opened read transaction for configuration data store, we need to mark it as stale.
113 * Latest read transaction will be allocated on another read using readLatestConfiguration
115 currentReadTxStale = true;
116 processDataChange(changes);
119 @SuppressWarnings("unchecked")
120 protected void removeData(final InstanceIdentifier<?> key, final Integer value) {
121 if (clazz.equals(key.getTargetType())) {
122 final InstanceIdentifier<Node> nodeIdent = key.firstIdentifierOf(Node.class);
123 Map<InstanceIdentifier<T>, Integer> map = null;
124 if (mapNodesForDelete.containsKey(nodeIdent)) {
125 map = mapNodesForDelete.get(nodeIdent);
128 map = new ConcurrentHashMap<>();
129 mapNodesForDelete.put(nodeIdent, map);
131 map.put((InstanceIdentifier<T>) key, value);
136 public void cleanForDisconnect(final InstanceIdentifier<Node> nodeIdent) {
137 mapNodesForDelete.remove(nodeIdent);
138 removedDataBetweenStatsCycle.remove(nodeIdent);
142 public void close() {
143 if (listenerRegistration != null) {
145 listenerRegistration.close();
146 } catch (final Exception e) {
147 LOG.error("Error by stop {} DataTreeChangeListener StatListeningCommiter.", clazz.getSimpleName(), e);
149 listenerRegistration = null;
156 * Method return actual DataObject identified by InstanceIdentifier from Config/DS
160 protected final <K extends DataObject> Optional<K> readLatestConfiguration(final InstanceIdentifier<K> path) {
161 for(int i = 0; i < 2; i++) {
162 boolean localReadTxStale = currentReadTxStale;
164 // This non-volatile read piggy backs the volatile currentReadTxStale read above to
165 // ensure visibility in case this method is called across threads (although not concurrently).
166 ReadOnlyTransaction localReadTx = currentReadTx;
167 if(localReadTx == null || localReadTxStale) {
168 if(localReadTx != null) {
172 localReadTx = dataBroker.newReadOnlyTransaction();
174 currentReadTx = localReadTx;
176 // Note - this volatile write also publishes the non-volatile currentReadTx write above.
177 currentReadTxStale = false;
181 return localReadTx.read(LogicalDatastoreType.CONFIGURATION, path).checkedGet();
182 } catch (final ReadFailedException e) {
183 LOG.debug("It wasn't possible to read {} from datastore. Exception: {}", path, e);
185 // Loop back and try again with a new Tx.
186 currentReadTxStale = true;
190 return Optional.absent();