opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/md/sal/binding/impl/AbstractForwardedTransaction.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 package org.opendaylight.controller.md.sal.binding.impl;
   9
  10 import java.util.ArrayList;
  11 import java.util.EnumMap;
  12 import java.util.Iterator;
  13 import java.util.List;
  14 import java.util.Map.Entry;
  15 import java.util.concurrent.ExecutionException;
  16 import java.util.concurrent.TimeUnit;
  17
  18 import javax.annotation.Nullable;
  19
  20 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  21 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  22 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  23 import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizationException;
  24 import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizationOperation;
  25 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadTransaction;
  26 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  27 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  28 import org.opendaylight.yangtools.concepts.Delegator;
  29 import org.opendaylight.yangtools.yang.binding.DataObject;
  30 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  31 import org.opendaylight.yangtools.yang.common.RpcResult;
  32 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier.PathArgument;
  33 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  34 import org.opendaylight.yangtools.yang.data.impl.codec.DeserializationException;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 import com.google.common.base.Function;
  39 import com.google.common.base.Optional;
  40 import com.google.common.cache.Cache;
  41 import com.google.common.cache.CacheBuilder;
  42 import com.google.common.util.concurrent.Futures;
  43 import com.google.common.util.concurrent.ListenableFuture;
  44
  45 public class AbstractForwardedTransaction<T extends AsyncTransaction<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, NormalizedNode<?, ?>>>
  46         implements Delegator<T> {
  47
  48     private static final Logger LOG = LoggerFactory.getLogger(AbstractForwardedTransaction.class);
  49     private final T delegate;
  50     private final static CacheBuilder<Object, Object> CACHE_BUILDER = CacheBuilder.newBuilder()
  51             .expireAfterWrite(10, TimeUnit.MILLISECONDS).maximumSize(100);
  52     private final BindingToNormalizedNodeCodec codec;
  53     private final EnumMap<LogicalDatastoreType, Cache<InstanceIdentifier<?>, DataObject>> cacheMap;
  54
  55     protected AbstractForwardedTransaction(final T delegate, final BindingToNormalizedNodeCodec codec) {
  56         super();
  57         this.delegate = delegate;
  58         this.codec = codec;
  59
  60         this.cacheMap = new EnumMap<>(LogicalDatastoreType.class);
  61         cacheMap.put(LogicalDatastoreType.OPERATIONAL, CACHE_BUILDER.<InstanceIdentifier<?>, DataObject> build());
  62         cacheMap.put(LogicalDatastoreType.CONFIGURATION, CACHE_BUILDER.<InstanceIdentifier<?>, DataObject> build());
  63
  64     }
  65
  66     @Override
  67     public T getDelegate() {
  68         return delegate;
  69     }
  70
  71     protected final BindingToNormalizedNodeCodec getCodec() {
  72         return codec;
  73     }
  74
  75     protected ListenableFuture<Optional<DataObject>> transformFuture(final LogicalDatastoreType store,
  76             final InstanceIdentifier<?> path, final ListenableFuture<Optional<NormalizedNode<?, ?>>> future) {
  77         return Futures.transform(future, new Function<Optional<NormalizedNode<?, ?>>, Optional<DataObject>>() {
  78             @Nullable
  79             @Override
  80             public Optional<DataObject> apply(@Nullable final Optional<NormalizedNode<?, ?>> normalizedNode) {
  81                 if (normalizedNode.isPresent()) {
  82                     final DataObject dataObject;
  83                     try {
  84                         dataObject = codec.toBinding(path, normalizedNode.get());
  85                     } catch (DeserializationException e) {
  86                         LOG.warn("Failed to create dataobject from node {}", normalizedNode.get(), e);
  87                         throw new IllegalStateException("Failed to create dataobject", e);
  88                     }
  89
  90                     if (dataObject != null) {
  91                         updateCache(store, path, dataObject);
  92                         return Optional.of(dataObject);
  93                     }
  94                 }
  95                 return Optional.absent();
  96             }
  97         });
  98     }
  99
 100     protected void doPut(final DOMDataWriteTransaction writeTransaction, final LogicalDatastoreType store,
 101             final InstanceIdentifier<?> path, final DataObject data) {
 102         invalidateCache(store, path);
 103         final Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, NormalizedNode<?, ?>> normalized = codec
 104                 .toNormalizedNode(path, data);
 105         writeTransaction.put(store, normalized.getKey(), normalized.getValue());
 106     }
 107
 108     protected void doPutWithEnsureParents(final DOMDataReadWriteTransaction writeTransaction,
 109             final LogicalDatastoreType store, final InstanceIdentifier<?> path, final DataObject data) {
 110         invalidateCache(store, path);
 111         final Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, NormalizedNode<?, ?>> normalized = codec
 112                 .toNormalizedNode(path, data);
 113
 114         final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier normalizedPath = normalized.getKey();
 115         ensureParentsByMerge(writeTransaction, store, normalizedPath, path);
 116         LOG.debug("Tx: {} : Putting data {}", getDelegate().getIdentifier(), normalizedPath);
 117         writeTransaction.put(store, normalizedPath, normalized.getValue());
 118     }
 119
 120     protected void doMergeWithEnsureParents(final DOMDataReadWriteTransaction writeTransaction,
 121             final LogicalDatastoreType store, final InstanceIdentifier<?> path, final DataObject data) {
 122         invalidateCache(store, path);
 123         final Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, NormalizedNode<?, ?>> normalized = codec
 124                 .toNormalizedNode(path, data);
 125
 126         final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier normalizedPath = normalized.getKey();
 127         ensureParentsByMerge(writeTransaction, store, normalizedPath, path);
 128         LOG.debug("Tx: {} : Merge data {}",getDelegate().getIdentifier(),normalizedPath);
 129         writeTransaction.merge(store, normalizedPath, normalized.getValue());
 130     }
 131
 132     private void ensureParentsByMerge(final DOMDataReadWriteTransaction writeTransaction,
 133             final LogicalDatastoreType store,
 134             final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier normalizedPath,
 135             final InstanceIdentifier<?> path) {
 136         List<PathArgument> currentArguments = new ArrayList<>();
 137         DataNormalizationOperation<?> currentOp = codec.getDataNormalizer().getRootOperation();
 138         Iterator<PathArgument> iterator = normalizedPath.getPath().iterator();
 139         while (iterator.hasNext()) {
 140             PathArgument currentArg = iterator.next();
 141             try {
 142                 currentOp = currentOp.getChild(currentArg);
 143             } catch (DataNormalizationException e) {
 144                 throw new IllegalArgumentException(String.format("Invalid child encountered in path %s", path), e);
 145             }
 146             currentArguments.add(currentArg);
 147             org.opendaylight.yangtools.yang.data.api.InstanceIdentifier currentPath = new org.opendaylight.yangtools.yang.data.api.InstanceIdentifier(
 148                     currentArguments);
 149
 150             final Optional<NormalizedNode<?, ?>> d;
 151             try {
 152                 d = writeTransaction.read(store, currentPath).get();
 153             } catch (InterruptedException | ExecutionException e) {
 154                 LOG.error("Failed to read pre-existing data from store {} path {}", store, currentPath, e);
 155                 throw new IllegalStateException("Failed to read pre-existing data", e);
 156             }
 157
 158             if (!d.isPresent() && iterator.hasNext()) {
 159                 writeTransaction.merge(store, currentPath, currentOp.createDefault(currentArg));
 160             }
 161         }
 162     }
 163
 164     protected void doMerge(final DOMDataWriteTransaction writeTransaction, final LogicalDatastoreType store,
 165             final InstanceIdentifier<?> path, final DataObject data) {
 166         invalidateCache(store, path);
 167         final Entry<org.opendaylight.yangtools.yang.data.api.InstanceIdentifier, NormalizedNode<?, ?>> normalized = codec
 168                 .toNormalizedNode(path, data);
 169         writeTransaction.merge(store, normalized.getKey(), normalized.getValue());
 170     }
 171
 172     protected void doDelete(final DOMDataWriteTransaction writeTransaction, final LogicalDatastoreType store,
 173             final InstanceIdentifier<?> path) {
 174         invalidateCache(store, path);
 175         final org.opendaylight.yangtools.yang.data.api.InstanceIdentifier normalized = codec.toNormalized(path);
 176         writeTransaction.delete(store, normalized);
 177     }
 178
 179     protected ListenableFuture<RpcResult<TransactionStatus>> doCommit(final DOMDataWriteTransaction writeTransaction) {
 180         return writeTransaction.commit();
 181     }
 182
 183     protected boolean doCancel(final DOMDataWriteTransaction writeTransaction) {
 184         return writeTransaction.cancel();
 185     }
 186
 187     protected ListenableFuture<Optional<DataObject>> doRead(final DOMDataReadTransaction readTransaction,
 188             final LogicalDatastoreType store, final InstanceIdentifier<?> path) {
 189         final DataObject dataObject = getFromCache(store, path);
 190         if (dataObject == null) {
 191             final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = readTransaction.read(store,
 192                     codec.toNormalized(path));
 193             return transformFuture(store, path, future);
 194         } else {
 195             return Futures.immediateFuture(Optional.of(dataObject));
 196         }
 197     }
 198
 199     private DataObject getFromCache(final LogicalDatastoreType store, final InstanceIdentifier<?> path) {
 200         Cache<InstanceIdentifier<?>, DataObject> cache = cacheMap.get(store);
 201         if (cache != null) {
 202             return cache.getIfPresent(path);
 203         }
 204         return null;
 205     }
 206
 207     private void updateCache(final LogicalDatastoreType store, final InstanceIdentifier<?> path,
 208             final DataObject dataObject) {
 209         // Check if cache exists. If not create one.
 210         Cache<InstanceIdentifier<?>, DataObject> cache = cacheMap.get(store);
 211         if (cache == null) {
 212             cache = CacheBuilder.newBuilder().maximumSize(1000).expireAfterWrite(1, TimeUnit.MINUTES).build();
 213
 214         }
 215
 216         cache.put(path, dataObject);
 217     }
 218
 219     private void invalidateCache(final LogicalDatastoreType store, final InstanceIdentifier<?> path) {
 220         // FIXME: Optimization: invalidate only parents and children of path
 221         Cache<InstanceIdentifier<?>, DataObject> cache = cacheMap.get(store);
 222         cache.invalidateAll();
 223         LOG.trace("Cache invalidated");
 224     }
 225
 226 }