data-change-counter/src/main/java/org/opendaylight/protocol/data/change/counter/TopologyDataChangeCounter.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.protocol.data.change.counter;
  10
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.MoreExecutors;
  15 import java.util.List;
  16 import java.util.concurrent.ExecutionException;
  17 import java.util.concurrent.atomic.AtomicBoolean;
  18 import java.util.concurrent.atomic.LongAdder;
  19 import org.opendaylight.mdsal.binding.api.DataBroker;
  20 import org.opendaylight.mdsal.binding.api.DataTreeChangeListener;
  21 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  22 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  23 import org.opendaylight.mdsal.binding.api.TransactionChain;
  24 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  25 import org.opendaylight.mdsal.common.api.CommitInfo;
  26 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.data.change.counter.rev160315.DataChangeCounter;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.data.change.counter.rev160315.data.change.counter.Counter;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.data.change.counter.rev160315.data.change.counter.CounterBuilder;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.data.change.counter.rev160315.data.change.counter.CounterKey;
  31 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NetworkTopology;
  32 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.TopologyId;
  33 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  34 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  35 import org.opendaylight.yangtools.concepts.Registration;
  36 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  37 import org.opendaylight.yangtools.yang.common.Empty;
  38 import org.opendaylight.yangtools.yang.common.Uint32;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 final class TopologyDataChangeCounter
  43         implements DataTreeChangeListener<Topology>, FutureCallback<Empty>, AutoCloseable {
  44     private static final Logger LOG = LoggerFactory.getLogger(TopologyDataChangeCounter.class);
  45
  46     private final DataBroker dataBroker;
  47     private final String counterId;
  48     private final InstanceIdentifier<Counter> counterInstanceId;
  49     private final LongAdder count = new LongAdder();
  50     private final AtomicBoolean closed = new AtomicBoolean(false);
  51     private final Registration registration;
  52
  53     private TransactionChain transactionChain;
  54
  55     TopologyDataChangeCounter(final DataBroker dataBroker, final String counterId, final String topologyName) {
  56         this.dataBroker = requireNonNull(dataBroker);
  57         transactionChain = dataBroker.createMergingTransactionChain();
  58         this.counterId = counterId;
  59         counterInstanceId = InstanceIdentifier.builder(DataChangeCounter.class)
  60                 .child(Counter.class, new CounterKey(this.counterId)).build();
  61         putCount(count.longValue());
  62         final InstanceIdentifier<Topology> topoIId = InstanceIdentifier.builder(NetworkTopology.class)
  63                 .child(Topology.class, new TopologyKey(new TopologyId(topologyName))).build();
  64         registration = dataBroker.registerTreeChangeListener(
  65             DataTreeIdentifier.of(LogicalDatastoreType.OPERATIONAL, topoIId), this);
  66         LOG.debug("Data change counter {} initiated", this.counterId);
  67         transactionChain.addCallback(this);
  68     }
  69
  70     @Override
  71     public synchronized void onDataTreeChanged(final List<DataTreeModification<Topology>> changes) {
  72         count.increment();
  73         final long inc = count.sum();
  74         LOG.debug("Data change #{} for counter {}", inc, counterId);
  75         putCount(inc);
  76     }
  77
  78     @Override
  79     public synchronized void close() {
  80         registration.close();
  81         final WriteTransaction wTx = dataBroker.newWriteOnlyTransaction();
  82         wTx.delete(LogicalDatastoreType.OPERATIONAL, counterInstanceId);
  83         try {
  84             wTx.commit().get();
  85         } catch (final ExecutionException | InterruptedException except) {
  86             LOG.warn("Error on remove data change counter {}", counterId, except);
  87         }
  88         transactionChain.close();
  89         LOG.debug("Data change counter {} removed", counterId);
  90     }
  91
  92     private void putCount(final long totalCount) {
  93         final WriteTransaction wTx = transactionChain.newWriteOnlyTransaction();
  94         final Counter counter = new CounterBuilder().setId(counterId).setCount(Uint32.valueOf(totalCount)).build();
  95         wTx.put(LogicalDatastoreType.OPERATIONAL, counterInstanceId, counter);
  96         wTx.commit().addCallback(new FutureCallback<CommitInfo>() {
  97             @Override
  98             public void onSuccess(final CommitInfo result) {
  99                 LOG.debug("Data change count update stored");
 100             }
 101
 102             @Override
 103             public void onFailure(final Throwable trw) {
 104                 LOG.error("Failed to store Data change count");
 105             }
 106         }, MoreExecutors.directExecutor());
 107     }
 108
 109     @Override
 110     public synchronized void onFailure(final Throwable cause) {
 111         LOG.warn("Transaction chain failure", cause);
 112         if (!closed.get()) {
 113             transactionChain.close();
 114             transactionChain = dataBroker.createMergingTransactionChain();
 115             transactionChain.addCallback(this);
 116         }
 117     }
 118
 119     @Override
 120     public synchronized void onSuccess(final Empty result) {
 121         LOG.debug("Transaction chain successful");
 122     }
 123 }