import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
-import java.util.Collection;
+import java.util.ArrayDeque;
import java.util.Deque;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
-import org.opendaylight.mdsal.dom.store.inmemory.DOMDataTreeShardProducer;
import org.opendaylight.mdsal.dom.store.inmemory.DOMDataTreeShardWriteTransaction;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.slf4j.Logger;
TransactionCommitFailedExceptionMapper.create("submit");
private static final AtomicLong COUNTER = new AtomicLong();
- private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> idToTransaction;
- private final Collection<YangInstanceIdentifier> childBoundaries;
+ private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions;
private final ShardedDOMDataTreeProducer producer;
+ private final ProducerLayout layout;
private final String identifier;
private final SettableFuture<Void> future = SettableFuture.create();
private DOMDataTreeWriteCursor openCursor;
ShardedDOMDataTreeWriteTransaction(final ShardedDOMDataTreeProducer producer,
- final Map<DOMDataTreeIdentifier, DOMDataTreeShardProducer> idToProducer,
- final Set<YangInstanceIdentifier> childRoots) {
+ final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions, final ProducerLayout layout) {
this.producer = Preconditions.checkNotNull(producer);
+ this.transactions = ImmutableMap.copyOf(transactions);
+ this.layout = Preconditions.checkNotNull(layout);
this.identifier = "SHARDED-DOM-" + COUNTER.getAndIncrement();
- idToTransaction = ImmutableMap.copyOf(Maps.transformValues(idToProducer,
- DOMDataTreeShardProducer::createTransaction));
- childBoundaries = Preconditions.checkNotNull(childRoots);
LOG.debug("Created new transaction {}", identifier);
}
private DOMDataTreeShardWriteTransaction lookup(final DOMDataTreeIdentifier prefix) {
- for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : idToTransaction.entrySet()) {
+ for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : transactions.entrySet()) {
if (e.getKey().contains(prefix)) {
Preconditions.checkArgument(!producer.isDelegatedToChild(prefix),
"Path %s is delegated to child producer.", prefix);
}
}
- throw new IllegalArgumentException(String.format("Path %s is not accessible from transaction %s",
- prefix, this));
+ return null;
}
@Override
if (openCursor != null) {
openCursor.close();
}
- for (final DOMDataTreeShardWriteTransaction tx : idToTransaction.values()) {
+ for (final DOMDataTreeShardWriteTransaction tx : transactions.values()) {
tx.close();
}
public synchronized DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
Preconditions.checkState(!closed, "Transaction is closed already");
Preconditions.checkState(openCursor == null, "There is still a cursor open");
+
final DOMDataTreeShardWriteTransaction lookup = lookup(prefix);
+ Preconditions.checkArgument(lookup != null, "Path %s is not accessible from transaction %s", prefix, this);
+
openCursor = new DelegatingCursor(lookup.createCursor(prefix), prefix);
return openCursor;
}
Preconditions.checkState(!closed, "Transaction %s is already closed", identifier);
Preconditions.checkState(openCursor == null, "Cannot submit transaction while there is a cursor open");
- producer.processTransaction(this);
+ producer.transactionSubmitted(this);
return submitFuture;
}
- CheckedFuture<Void, TransactionCommitFailedException> doSubmit(
- final Consumer<ShardedDOMDataTreeWriteTransaction> success,
+ void doSubmit(final Consumer<ShardedDOMDataTreeWriteTransaction> success,
final BiConsumer<ShardedDOMDataTreeWriteTransaction, Throwable> failure) {
final ListenableFuture<List<Void>> listListenableFuture = Futures.allAsList(
- idToTransaction.values().stream().map(tx -> {
+ transactions.values().stream().map(tx -> {
LOG.debug("Readying tx {}", identifier);
tx.ready();
return tx.submit();
}).collect(Collectors.toList()));
- final SettableFuture<Void> ret = SettableFuture.create();
Futures.addCallback(listListenableFuture, new FutureCallback<List<Void>>() {
@Override
public void onSuccess(final List<Void> result) {
success.accept(ShardedDOMDataTreeWriteTransaction.this);
- ret.set(null);
}
@Override
public void onFailure(final Throwable exp) {
failure.accept(ShardedDOMDataTreeWriteTransaction.this, exp);
- ret.setException(exp);
}
});
-
- return Futures.makeChecked(ret, SUBMIT_FAILED_MAPPER);
}
void onTransactionSuccess(final Void result) {
}
private class DelegatingCursor implements DOMDataTreeWriteCursor {
-
+ private final Deque<PathArgument> path = new ArrayDeque<>();
private final DOMDataTreeWriteCursor delegate;
private final DOMDataTreeIdentifier rootPosition;
- private final Deque<PathArgument> path = new LinkedList<>();
DelegatingCursor(final DOMDataTreeWriteCursor delegate, final DOMDataTreeIdentifier rootPosition) {
this.delegate = Preconditions.checkNotNull(delegate);
@Override
public void enter(@Nonnull final PathArgument child) {
checkAvailable(child);
- path.push(child);
delegate.enter(child);
+ path.push(child);
}
@Override
@Override
public void exit() {
- path.pop();
delegate.exit();
+ path.pop();
}
@Override
public void exit(final int depth) {
+ delegate.exit(depth);
for (int i = 0; i < depth; i++) {
path.pop();
}
- delegate.exit(depth);
}
@Override
}
void checkAvailable(final PathArgument child) {
- path.add(child);
- final YangInstanceIdentifier yid = YangInstanceIdentifier.create(path);
- childBoundaries.forEach(id -> {
- if (id.contains(yid)) {
- path.removeLast();
- throw new IllegalArgumentException("Path {" + yid + "} is not available to this cursor"
- + " since it's already claimed by a child producer");
- }
- });
- path.removeLast();
+ layout.checkAvailable(path, child);
}
}
+
+ ProducerLayout getLayout() {
+ return layout;
+ }
}