dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/AbstractDOMShardTreeChangePublisher.java

   1 /*
   2  * Copyright (c) 2016 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.mdsal.dom.store.inmemory;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import java.util.ArrayList;
  14 import java.util.Collection;
  15 import java.util.Collections;
  16 import java.util.HashMap;
  17 import java.util.Iterator;
  18 import java.util.List;
  19 import java.util.Map;
  20 import java.util.Optional;
  21 import java.util.stream.Collectors;
  22 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
  23 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  24 import org.opendaylight.mdsal.dom.spi.AbstractDOMDataTreeChangeListenerRegistration;
  25 import org.opendaylight.mdsal.dom.spi.RegistrationTreeNode;
  26 import org.opendaylight.mdsal.dom.spi.shard.ChildShardContext;
  27 import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTreeChangePublisher;
  28 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTreeChangePublisher;
  29 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  30 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  31 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  32 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  33 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerNode;
  34 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  36 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  37 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  38 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
  39 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNodes;
  40 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
  41 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  42 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
  43 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.NormalizedNodeBuilder;
  44 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  45 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapEntryNodeBuilder;
  46 import org.slf4j.Logger;
  47 import org.slf4j.LoggerFactory;
  48
  49 abstract class AbstractDOMShardTreeChangePublisher extends AbstractDOMStoreTreeChangePublisher {
  50
  51     private static final Logger LOG = LoggerFactory.getLogger(AbstractDOMShardTreeChangePublisher.class);
  52
  53     private final YangInstanceIdentifier shardPath;
  54     private final Map<DOMDataTreeIdentifier, ChildShardContext> childShards;
  55     private final DataTree dataTree;
  56
  57     protected AbstractDOMShardTreeChangePublisher(final DataTree dataTree,
  58             final YangInstanceIdentifier shardPath, final Map<DOMDataTreeIdentifier, ChildShardContext> childShards) {
  59         this.dataTree = requireNonNull(dataTree);
  60         this.shardPath = requireNonNull(shardPath);
  61         this.childShards = requireNonNull(childShards);
  62     }
  63
  64     @Override
  65     public <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
  66             registerTreeChangeListener(final YangInstanceIdentifier path, final L listener) {
  67         takeLock();
  68         try {
  69             return setupListenerContext(path, listener);
  70         } finally {
  71             releaseLock();
  72         }
  73     }
  74
  75     private <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
  76             setupListenerContext(final YangInstanceIdentifier listenerPath, final L listener) {
  77         // we need to register the listener registration path based on the shards root
  78         // we have to strip the shard path from the listener path and then register
  79         YangInstanceIdentifier strippedIdentifier = listenerPath;
  80         if (!shardPath.isEmpty()) {
  81             strippedIdentifier = YangInstanceIdentifier.create(stripShardPath(listenerPath));
  82         }
  83
  84         final DOMDataTreeListenerWithSubshards subshardListener =
  85                 new DOMDataTreeListenerWithSubshards(dataTree, strippedIdentifier, listener);
  86         final AbstractDOMDataTreeChangeListenerRegistration<L> reg =
  87                 setupContextWithoutSubshards(strippedIdentifier, subshardListener);
  88
  89         for (final ChildShardContext maybeAffected : childShards.values()) {
  90             if (listenerPath.contains(maybeAffected.getPrefix().getRootIdentifier())) {
  91                 // consumer has initialDataChangeEvent subshard somewhere on lower level
  92                 // register to the notification manager with snapshot and forward child notifications to parent
  93                 LOG.debug("Adding new subshard{{}} to listener at {}", maybeAffected.getPrefix(), listenerPath);
  94                 subshardListener.addSubshard(maybeAffected);
  95             } else if (maybeAffected.getPrefix().getRootIdentifier().contains(listenerPath)) {
  96                 // bind path is inside subshard
  97                 // TODO can this happen? seems like in ShardedDOMDataTree we are
  98                 // already registering to the lowest shard possible
  99                 throw new UnsupportedOperationException("Listener should be registered directly "
 100                         + "into initialDataChangeEvent subshard");
 101             }
 102         }
 103
 104         initialDataChangeEvent(listenerPath, listener);
 105
 106         return reg;
 107     }
 108
 109     private <L extends DOMDataTreeChangeListener> void initialDataChangeEvent(
 110             final YangInstanceIdentifier listenerPath, final L listener) {
 111         // FIXME Add support for wildcard listeners
 112         final Optional<NormalizedNode<?, ?>> preExistingData = dataTree.takeSnapshot()
 113                 .readNode(YangInstanceIdentifier.create(stripShardPath(listenerPath)));
 114         final DataTreeCandidate initialCandidate;
 115
 116         if (preExistingData.isPresent()) {
 117             final NormalizedNode<?, ?> data = preExistingData.get();
 118             checkState(data instanceof DataContainerNode, "Expected DataContainer node, but was {}", data.getClass());
 119             // if we are listening on root of some shard we still get
 120             // empty normalized node, root is always present
 121             if (((DataContainerNode<?>) data).getValue().isEmpty()) {
 122                 initialCandidate = DataTreeCandidates.newDataTreeCandidate(listenerPath,
 123                     DataTreeCandidateNodes.empty(data.getIdentifier()));
 124             } else {
 125                 initialCandidate = DataTreeCandidates.fromNormalizedNode(listenerPath,
 126                     translateRootShardIdentifierToListenerPath(listenerPath, preExistingData.get()));
 127             }
 128         } else {
 129             initialCandidate = DataTreeCandidates.newDataTreeCandidate(listenerPath,
 130                 DataTreeCandidateNodes.empty(listenerPath.getLastPathArgument()));
 131         }
 132
 133         listener.onDataTreeChanged(Collections.singleton(initialCandidate));
 134     }
 135
 136     private static NormalizedNode<?, ?> translateRootShardIdentifierToListenerPath(
 137             final YangInstanceIdentifier listenerPath, final NormalizedNode<?, ?> node) {
 138         if (listenerPath.isEmpty()) {
 139             return node;
 140         }
 141
 142         final NormalizedNodeBuilder nodeBuilder;
 143         if (node instanceof ContainerNode) {
 144             nodeBuilder = ImmutableContainerNodeBuilder.create().withValue(((ContainerNode) node).getValue());
 145         } else if (node instanceof MapEntryNode) {
 146             nodeBuilder = ImmutableMapEntryNodeBuilder.create().withValue(((MapEntryNode) node).getValue());
 147         } else {
 148             throw new IllegalArgumentException("Expected ContainerNode or MapEntryNode, but was " + node.getClass());
 149         }
 150         nodeBuilder.withNodeIdentifier(listenerPath.getLastPathArgument());
 151         return nodeBuilder.build();
 152     }
 153
 154     private <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
 155             setupContextWithoutSubshards(final YangInstanceIdentifier listenerPath,
 156                     final DOMDataTreeListenerWithSubshards listener) {
 157         LOG.debug("Registering root listener at {}", listenerPath);
 158         final RegistrationTreeNode<AbstractDOMDataTreeChangeListenerRegistration<?>> node =
 159                 findNodeFor(listenerPath.getPathArguments());
 160         @SuppressWarnings("unchecked")
 161         final AbstractDOMDataTreeChangeListenerRegistration<L> registration =
 162                 new AbstractDOMDataTreeChangeListenerRegistration<L>((L) listener) {
 163             @Override
 164             protected void removeRegistration() {
 165                 listener.close();
 166                 AbstractDOMShardTreeChangePublisher.this.removeRegistration(node, this);
 167                 registrationRemoved(this);
 168             }
 169         };
 170         addRegistration(node, registration);
 171         return registration;
 172     }
 173
 174     private Iterable<PathArgument> stripShardPath(final YangInstanceIdentifier listenerPath) {
 175         if (shardPath.isEmpty()) {
 176             return listenerPath.getPathArguments();
 177         }
 178
 179         final List<PathArgument> listenerPathArgs = new ArrayList<>(listenerPath.getPathArguments());
 180         final Iterator<PathArgument> shardIter = shardPath.getPathArguments().iterator();
 181         final Iterator<PathArgument> listenerIter = listenerPathArgs.iterator();
 182
 183         while (shardIter.hasNext()) {
 184             if (shardIter.next().equals(listenerIter.next())) {
 185                 listenerIter.remove();
 186             } else {
 187                 break;
 188             }
 189         }
 190
 191         return listenerPathArgs;
 192     }
 193
 194     private static final class DOMDataTreeListenerWithSubshards implements DOMDataTreeChangeListener {
 195
 196         private final DataTree dataTree;
 197         private final YangInstanceIdentifier listenerPath;
 198         private final DOMDataTreeChangeListener delegate;
 199
 200         private final Map<YangInstanceIdentifier, ListenerRegistration<DOMDataTreeChangeListener>> registrations =
 201                 new HashMap<>();
 202
 203         DOMDataTreeListenerWithSubshards(final DataTree dataTree, final YangInstanceIdentifier listenerPath,
 204                 final DOMDataTreeChangeListener delegate) {
 205             this.dataTree = requireNonNull(dataTree);
 206             this.listenerPath = requireNonNull(listenerPath);
 207             this.delegate = requireNonNull(delegate);
 208         }
 209
 210         @Override
 211         public void onDataTreeChanged(final Collection<DataTreeCandidate> changes) {
 212             LOG.debug("Received data changed {}", changes);
 213             delegate.onDataTreeChanged(changes);
 214         }
 215
 216         void onDataTreeChanged(final YangInstanceIdentifier rootPath, final Collection<DataTreeCandidate> changes) {
 217             final List<DataTreeCandidate> newCandidates = changes.stream()
 218                     .map(candidate -> DataTreeCandidates.newDataTreeCandidate(rootPath, candidate.getRootNode()))
 219                     .collect(Collectors.toList());
 220             delegate.onDataTreeChanged(Collections.singleton(applyChanges(newCandidates)));
 221         }
 222
 223         void addSubshard(final ChildShardContext context) {
 224             checkState(context.getShard() instanceof DOMStoreTreeChangePublisher,
 225                     "All subshards that are initialDataChangeEvent part of ListenerContext need to be listenable");
 226
 227             final DOMStoreTreeChangePublisher listenableShard = (DOMStoreTreeChangePublisher) context.getShard();
 228             // since this is going into subshard we want to listen for ALL changes in the subshard
 229             registrations.put(context.getPrefix().getRootIdentifier(),
 230                 listenableShard.registerTreeChangeListener(
 231                         context.getPrefix().getRootIdentifier(), changes -> onDataTreeChanged(
 232                                 context.getPrefix().getRootIdentifier(), changes)));
 233         }
 234
 235         void close() {
 236             for (final ListenerRegistration<DOMDataTreeChangeListener> registration : registrations.values()) {
 237                 registration.close();
 238             }
 239             registrations.clear();
 240         }
 241
 242         private DataTreeCandidate applyChanges(final Collection<DataTreeCandidate> changes) {
 243             final DataTreeModification modification = dataTree.takeSnapshot().newModification();
 244             for (final DataTreeCandidate change : changes) {
 245                 DataTreeCandidates.applyToModification(modification, change);
 246             }
 247
 248             modification.ready();
 249             try {
 250                 dataTree.validate(modification);
 251             } catch (final DataValidationFailedException e) {
 252                 LOG.error("Validation failed for built modification", e);
 253                 throw new RuntimeException("Notification validation failed", e);
 254             }
 255
 256             // strip nodes we do not need since this listener doesn't have to be registered at the root of the DataTree
 257             DataTreeCandidateNode modifiedChild = dataTree.prepare(modification).getRootNode();
 258             for (final PathArgument pathArgument : listenerPath.getPathArguments()) {
 259                 modifiedChild = modifiedChild.getModifiedChild(pathArgument);
 260             }
 261
 262             if (modifiedChild == null) {
 263                 modifiedChild = DataTreeCandidateNodes.empty(listenerPath.getLastPathArgument());
 264             }
 265
 266             return DataTreeCandidates.newDataTreeCandidate(listenerPath, modifiedChild);
 267         }
 268     }
 269 }