mdsalutil/mdsalutil-api/src/main/java/org/opendaylight/genius/mdsalutil/cache/DataObjectCache.java

   1 /*
   2  * Copyright (c) 2017 Inocybe Technologies 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 package org.opendaylight.genius.mdsalutil.cache;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.cache.CacheBuilder;
  13 import com.google.common.cache.CacheLoader;
  14 import com.google.common.cache.LoadingCache;
  15 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  16 import java.util.Collection;
  17 import java.util.Objects;
  18 import java.util.Optional;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.atomic.AtomicBoolean;
  21 import java.util.function.BiFunction;
  22 import java.util.function.Function;
  23 import java.util.stream.Collectors;
  24 import java.util.stream.Stream;
  25 import javax.annotation.PreDestroy;
  26 import org.eclipse.jdt.annotation.NonNull;
  27 import org.opendaylight.genius.datastoreutils.SingleTransactionDataBroker;
  28 import org.opendaylight.infrautils.caches.CacheProvider;
  29 import org.opendaylight.mdsal.binding.api.ClusteredDataTreeChangeListener;
  30 import org.opendaylight.mdsal.binding.api.DataBroker;
  31 import org.opendaylight.mdsal.binding.api.DataObjectModification;
  32 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  33 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  34 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  35 import org.opendaylight.mdsal.common.api.ReadFailedException;
  36 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  37 import org.opendaylight.yangtools.yang.binding.DataObject;
  38 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 /**
  43  * Caches DataObjects of a particular type. The cache is updated by a DataTreeChangeListener.
  44  *
  45  * @author Thomas Pantelis
  46  */
  47 public class DataObjectCache<K, V extends DataObject> implements AutoCloseable {
  48
  49     private static final Logger LOG = LoggerFactory.getLogger(DataObjectCache.class);
  50
  51     private final SingleTransactionDataBroker broker;
  52     private final LoadingCache<K, Optional<V>> cache;
  53     private final AtomicBoolean isClosed = new AtomicBoolean();
  54     protected ListenerRegistration<?> listenerRegistration;
  55     protected ClusteredDataTreeChangeListener<V> dataObjectListener;
  56
  57     /**
  58      * Constructor.
  59      *
  60      * @param dataObjectClass the DataObject class to cache
  61      * @param dataBroker the DataBroker
  62      * @param datastoreType the LogicalDatastoreType
  63      * @param cacheProvider the CacheProvider used to instantiate the Cache
  64      * @param keyFunction the function used to convert or extract the key instance on change notification
  65      * @param instanceIdFunction the function used to convert a key instance to an InstanceIdentifier on read
  66      */
  67     public DataObjectCache(Class<V> dataObjectClass, DataBroker dataBroker, LogicalDatastoreType datastoreType,
  68                            InstanceIdentifier<V> listetenerRegistrationPath, CacheProvider cacheProvider,
  69                            BiFunction<InstanceIdentifier<V>, V, K> keyFunction,
  70                            Function<K, InstanceIdentifier<V>> instanceIdFunction) {
  71         this(dataObjectClass, dataBroker, datastoreType, cacheProvider, keyFunction, instanceIdFunction);
  72         listenerRegistration = dataBroker.registerDataTreeChangeListener(DataTreeIdentifier.create(
  73                 datastoreType, listetenerRegistrationPath), dataObjectListener);
  74
  75     }
  76
  77     public DataObjectCache(Class<V> dataObjectClass, DataBroker dataBroker, LogicalDatastoreType datastoreType,
  78                            CacheProvider cacheProvider, BiFunction<InstanceIdentifier<V>, V, K> keyFunction,
  79                            Function<K, InstanceIdentifier<V>> instanceIdFunction) {
  80         Objects.requireNonNull(keyFunction);
  81         Objects.requireNonNull(instanceIdFunction);
  82         this.broker = new SingleTransactionDataBroker(Objects.requireNonNull(dataBroker));
  83
  84         requireNonNull(cacheProvider, "cacheProvider");
  85         cache = CacheBuilder.newBuilder().build(new CacheLoader<K, Optional<V>>() {
  86             @Override
  87             public Optional<V> load(K key) throws ReadFailedException, ExecutionException, InterruptedException  {
  88                 return broker.syncReadOptional(datastoreType, instanceIdFunction.apply(key));
  89             }
  90         });
  91
  92         dataObjectListener = changes -> {
  93             for (DataTreeModification<V> dataTreeModification : changes) {
  94                 DataObjectModification<V> rootNode = dataTreeModification.getRootNode();
  95                 InstanceIdentifier<V> path = dataTreeModification.getRootPath().getRootIdentifier();
  96                 switch (rootNode.getModificationType()) {
  97                     case WRITE:
  98                     case SUBTREE_MODIFIED:
  99                         V dataAfter = rootNode.getDataAfter();
 100                         cache.put(keyFunction.apply(path, dataAfter), Optional.ofNullable(dataAfter));
 101                         added(path, dataAfter);
 102                         break;
 103                     case DELETE:
 104                         V dataBefore = rootNode.getDataBefore();
 105                         cache.invalidate(keyFunction.apply(path, dataBefore));
 106                         removed(path, dataBefore);
 107                         break;
 108                     default:
 109                         break;
 110                 }
 111             }
 112         };
 113     }
 114
 115     @Override
 116     @PreDestroy
 117     public void close() {
 118         if (isClosed.compareAndSet(false, true)) {
 119             if (listenerRegistration != null) {
 120                 listenerRegistration.close();
 121             }
 122             cache.cleanUp();
 123         } else {
 124             LOG.warn("Lifecycled object already closed; ignoring extra close()");
 125         }
 126     }
 127
 128     protected void checkIsClosed() throws ReadFailedException {
 129         if (isClosed.get()) {
 130             throw new ReadFailedException("Lifecycled object is already closed: " + this.toString());
 131         }
 132     }
 133
 134     /**
 135      * Gets the DataObject for the given key. If there's no DataObject cached, it will be read from the data store
 136      * and put in the cache if it exists.
 137      *
 138      * @param key identifies the DataObject to query
 139      * @return if the data for the supplied key exists, returns an Optional object containing the data; otherwise,
 140      *         returns Optional#absent()
 141      * @throws ReadFailedException if that data isn't cached and the read to fetch it fails
 142      */
 143     @NonNull
 144     // The ExecutionException cause should be a ReadFailedException - ok to cast.
 145     @SuppressFBWarnings("BC_UNCONFIRMED_CAST_OF_RETURN_VALUE")
 146     @SuppressWarnings("checkstyle:AvoidHidingCauseException")
 147     public Optional<V> get(@NonNull K key) throws ReadFailedException {
 148         checkIsClosed();
 149         try {
 150             return cache.get(key);
 151         } catch (ExecutionException e) {
 152             throw (ReadFailedException) e.getCause();
 153         }
 154     }
 155
 156     /**
 157      * Gets all DataObjects currently in the cache.
 158      *
 159      * @return the DataObjects currently in the cache
 160      */
 161     @NonNull
 162     public Collection<V> getAllPresent() {
 163         return cache.asMap().values().stream().flatMap(optional -> optional.isPresent()
 164                 ? Stream.of(optional.get()) : Stream.empty()).collect(Collectors.toList());
 165     }
 166
 167     protected void added(InstanceIdentifier<V> path, V dataObject) {
 168     }
 169
 170     protected void removed(InstanceIdentifier<V> path, V dataObject) {
 171     }
 172
 173 }