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.Collection;
16 import java.util.concurrent.Callable;
17 import java.util.concurrent.ConcurrentHashMap;
19 import org.opendaylight.controller.md.sal.binding.api.*;
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.openflowplugin.common.wait.SimpleTaskRetryLooper;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
29 import org.opendaylight.yangtools.concepts.ListenerRegistration;
30 import org.opendaylight.yangtools.yang.binding.DataObject;
31 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
32 import org.opendaylight.yangtools.yang.binding.NotificationListener;
34 import org.slf4j.Logger;
35 import org.slf4j.LoggerFactory;
39 * org.opendaylight.openflowplugin.applications.statistics.manager.impl
41 * StatAbstractListeneningCommiter
42 * Class is abstract implementation for all Configuration/DataStore DataChange
43 * listenable DataObjects like flows, groups, meters. It is a holder for common
44 * functionality needed by construction/destruction class and for DataChange
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<>();
58 private final Class<T> clazz;
60 private final DataBroker dataBroker;
62 protected final StatNodeRegistration nodeRegistrationManager;
64 private ReadOnlyTransaction currentReadTx;
65 private volatile boolean currentReadTxStale;
67 private static final int STARTUP_LOOP_TICK = 500;
68 private static final int STARTUP_LOOP_MAX_RETRIES = 8;
70 private final DataTreeIdentifier<T> treeId =
71 new DataTreeIdentifier<>(LogicalDatastoreType.CONFIGURATION, getWildCardedRegistrationPath());
73 /* Constructor has to make a registration */
74 public StatAbstractListenCommit(final StatisticsManager manager, final DataBroker db,
75 final NotificationProviderService nps, final Class<T> clazz, final StatNodeRegistration nodeRegistrationManager) {
76 super(manager,nps, nodeRegistrationManager);
77 this.clazz = Preconditions.checkNotNull(clazz, "Referenced Class can not be null");
78 Preconditions.checkArgument(db != null, "DataBroker can not be null!");
80 this.nodeRegistrationManager = nodeRegistrationManager;
82 SimpleTaskRetryLooper looper = new SimpleTaskRetryLooper(STARTUP_LOOP_TICK, STARTUP_LOOP_MAX_RETRIES);
84 listenerRegistration = looper.loopUntilNoException(new Callable<ListenerRegistration<StatAbstractListenCommit<T, N>>>() {
86 public ListenerRegistration<StatAbstractListenCommit<T, N>> call() throws Exception {
87 return db.registerDataTreeChangeListener(treeId,StatAbstractListenCommit.this);
90 }catch(final Exception ex){
91 LOG.debug(" StatAbstractListenCommit DataChange listener registration failed {}", ex.getMessage());
92 throw new IllegalStateException("Notification supplier startup fail! System needs restart.", ex);
97 * Method returns WildCarded Path which is used for registration as a listening path changes in
98 * {@link org.opendaylight.controller.md.sal.binding.api.DataTreeChangeListener}
101 protected abstract InstanceIdentifier<T> getWildCardedRegistrationPath();
104 public void onDataTreeChanged(Collection<DataTreeModification<T>> changes) {
105 Preconditions.checkNotNull(changes, "Changes must not be null!");
107 * If we have opened read transaction for configuration data store, we need to mark it as stale.
109 * Latest read transaction will be allocated on another read using readLatestConfiguration
111 currentReadTxStale = true;
114 @SuppressWarnings("unchecked")
115 protected void removeData(final InstanceIdentifier<?> key, final Integer value) {
116 if (clazz.equals(key.getTargetType())) {
117 final InstanceIdentifier<Node> nodeIdent = key.firstIdentifierOf(Node.class);
118 Map<InstanceIdentifier<T>, Integer> map = null;
119 if (mapNodesForDelete.containsKey(nodeIdent)) {
120 map = mapNodesForDelete.get(nodeIdent);
123 map = new ConcurrentHashMap<>();
124 mapNodesForDelete.put(nodeIdent, map);
126 map.put((InstanceIdentifier<T>) key, value);
131 public void cleanForDisconnect(final InstanceIdentifier<Node> nodeIdent) {
132 mapNodesForDelete.remove(nodeIdent);
136 public void close() {
137 if (listenerRegistration != null) {
139 listenerRegistration.close();
140 } catch (final Exception e) {
141 LOG.error("Error by stop {} DataChange StatListeningCommiter.", clazz.getSimpleName(), e);
143 listenerRegistration = null;
150 * Method return actual DataObject identified by InstanceIdentifier from Config/DS
154 protected final <K extends DataObject> Optional<K> readLatestConfiguration(final InstanceIdentifier<K> path) {
155 for(int i = 0; i < 2; i++) {
156 boolean localReadTxStale = currentReadTxStale;
158 // This non-volatile read piggy backs the volatile currentReadTxStale read above to
159 // ensure visibility in case this method is called across threads (although not concurrently).
160 ReadOnlyTransaction localReadTx = currentReadTx;
161 if(localReadTx == null || localReadTxStale) {
162 if(localReadTx != null) {
166 localReadTx = dataBroker.newReadOnlyTransaction();
168 currentReadTx = localReadTx;
170 // Note - this volatile write also publishes the non-volatile currentReadTx write above.
171 currentReadTxStale = false;
175 return localReadTx.read(LogicalDatastoreType.CONFIGURATION, path).checkedGet();
176 } catch (final ReadFailedException e) {
177 LOG.debug("It wasn't possible to read {} from datastore. Exception: {}", path, e);
179 // Loop back and try again with a new Tx.
180 currentReadTxStale = true;
184 return Optional.absent();