2 * Copyright (c) 2016 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
8 package org.opendaylight.mdsal.dom.broker;
10 import com.google.common.base.Preconditions;
11 import com.google.common.collect.Lists;
12 import com.google.common.util.concurrent.FluentFuture;
13 import com.google.common.util.concurrent.FutureCallback;
14 import com.google.common.util.concurrent.MoreExecutors;
15 import com.google.common.util.concurrent.SettableFuture;
16 import java.util.Collection;
17 import java.util.Collections;
18 import java.util.List;
20 import java.util.Optional;
21 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
22 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
23 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
24 import org.opendaylight.mdsal.dom.api.DOMDataTreeListeningException;
25 import org.opendaylight.mdsal.dom.api.DOMDataTreeLoopException;
26 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
27 import org.opendaylight.mdsal.dom.api.DOMDataTreeService;
28 import org.opendaylight.yangtools.concepts.ListenerRegistration;
29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
30 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
31 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
32 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
33 import org.slf4j.Logger;
34 import org.slf4j.LoggerFactory;
36 public class ShardedDOMReadTransactionAdapter implements DOMDataTreeReadTransaction {
38 private static final Logger LOG = LoggerFactory.getLogger(ShardedDOMReadTransactionAdapter.class.getName());
40 private final List<ListenerRegistration<DOMDataTreeListener>> registrations = Lists.newArrayList();
41 private final DOMDataTreeService service;
42 private final Object txIdentifier;
44 private boolean finished = false;
46 ShardedDOMReadTransactionAdapter(final Object identifier, final DOMDataTreeService service) {
47 this.service = Preconditions.checkNotNull(service);
48 this.txIdentifier = Preconditions.checkNotNull(identifier);
53 // TODO should we also cancel all read futures?
54 LOG.debug("{}: Closing read transaction", txIdentifier);
59 registrations.forEach(ListenerRegistration::close);
64 public Object getIdentifier() {
69 public FluentFuture<Optional<NormalizedNode<?, ?>>> read(final LogicalDatastoreType store,
70 final YangInstanceIdentifier path) {
72 LOG.debug("{}: Invoking read at {}:{}", txIdentifier, store, path);
73 final ListenerRegistration<DOMDataTreeListener> reg;
74 final SettableFuture<Optional<NormalizedNode<?, ?>>> initialDataTreeChangeFuture = SettableFuture.create();
76 reg = service.registerListener(new ReadShardedListener(initialDataTreeChangeFuture),
77 Collections.singleton(new DOMDataTreeIdentifier(store, path)), false, Collections.emptyList());
78 registrations.add(reg);
79 } catch (final DOMDataTreeLoopException e) {
80 // This should not happen, we are not specifying any
81 // producers when registering listener
82 throw new IllegalStateException("Loop in listener and producers detected", e);
85 // After data tree change future is finished, we can close the listener registration
86 initialDataTreeChangeFuture.addCallback(new FutureCallback<Optional<NormalizedNode<?, ?>>>() {
88 public void onSuccess(final Optional<NormalizedNode<?, ?>> result) {
93 public void onFailure(final Throwable throwable) {
96 }, MoreExecutors.directExecutor());
98 return initialDataTreeChangeFuture;
102 public FluentFuture<Boolean> exists(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
104 LOG.debug("{}: Invoking exists at {}:{}", txIdentifier, store, path);
105 return read(store, path).transform(Optional::isPresent, MoreExecutors.directExecutor());
108 private void checkRunning() {
109 Preconditions.checkState(!finished, "Transaction is already closed");
112 static class ReadShardedListener implements DOMDataTreeListener {
114 private final SettableFuture<Optional<NormalizedNode<?, ?>>> readResultFuture;
116 ReadShardedListener(final SettableFuture<Optional<NormalizedNode<?, ?>>> future) {
117 this.readResultFuture = Preconditions.checkNotNull(future);
121 public void onDataTreeChanged(final Collection<DataTreeCandidate> changes,
122 final Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>> subtrees) {
123 Preconditions.checkState(changes.size() == 1 && subtrees.size() == 1,
124 "DOMDataTreeListener registered exactly on one subtree");
126 for (final DataTreeCandidate change : changes) {
127 if (change.getRootNode().getModificationType().equals(ModificationType.UNMODIFIED)) {
128 readResultFuture.set(Optional.empty());
133 for (final NormalizedNode<?, ?> initialState : subtrees.values()) {
134 readResultFuture.set(Optional.of(initialState));
139 public void onDataTreeFailed(final Collection<DOMDataTreeListeningException> causes) {
140 // TODO If we get just one exception, we don't need to do
143 // We chain all exceptions and return aggregated one
144 readResultFuture.setException(new DOMDataTreeListeningException("Aggregated DOMDataTreeListening exception",
145 causes.stream().reduce((e1, e2) -> {
146 e1.addSuppressed(e2);