dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InmemoryDOMDataTreeShardWriteTransaction.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
   9 package org.opendaylight.mdsal.dom.store.inmemory;
  10
  11 import com.google.common.base.Optional;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.util.concurrent.CheckedFuture;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import com.google.common.util.concurrent.ListeningExecutorService;
  16 import com.google.common.util.concurrent.MoreExecutors;
  17 import java.util.ArrayList;
  18 import java.util.Iterator;
  19 import java.util.Map.Entry;
  20 import java.util.concurrent.Callable;
  21 import java.util.concurrent.Executors;
  22 import org.opendaylight.mdsal.common.api.ReadFailedException;
  23 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  24 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  25 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
  26 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  28 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  29 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  30 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 class InmemoryDOMDataTreeShardWriteTransaction implements DOMDataTreeShardWriteTransaction {
  35
  36     private static final Logger LOG = LoggerFactory.getLogger(InmemoryDOMDataTreeShardWriteTransaction.class);
  37
  38     private enum SimpleCursorOperation {
  39         MERGE {
  40             @Override
  41             void applyOnLeaf(final DOMDataTreeWriteCursor cursor, final PathArgument child,
  42                     final NormalizedNode<?, ?> data) {
  43                 cursor.merge(child, data);
  44             }
  45         },
  46         DELETE {
  47             @Override
  48             void applyOnLeaf(final DOMDataTreeWriteCursor cursor, final PathArgument child,
  49                     final NormalizedNode<?, ?> data) {
  50                 cursor.delete(child);
  51             }
  52         },
  53         WRITE {
  54             @Override
  55             void applyOnLeaf(final DOMDataTreeWriteCursor cursor, final PathArgument child,
  56                     final NormalizedNode<?, ?> data) {
  57                 cursor.write(child, data);
  58             }
  59         };
  60
  61         abstract void applyOnLeaf(DOMDataTreeWriteCursor cursor, PathArgument child, NormalizedNode<?, ?> data);
  62
  63         void apply(final DOMDataTreeWriteCursor cursor, final YangInstanceIdentifier path,
  64                 final NormalizedNode<?, ?> data) {
  65             int enterCount = 0;
  66             Iterator<PathArgument> it = path.getPathArguments().iterator();
  67             while (it.hasNext()) {
  68                 PathArgument currentArg = it.next();
  69                 if (it.hasNext()) {
  70                     // We need to enter one level deeper, we are not at leaf (modified) node
  71                     cursor.enter(currentArg);
  72                     enterCount++;
  73                 } else {
  74                     applyOnLeaf(cursor, currentArg, data);
  75                 }
  76             }
  77             cursor.exit(enterCount);
  78         }
  79     }
  80
  81     private final ShardDataModification modification;
  82     private DOMDataTreeWriteCursor cursor;
  83     private DataTree rootShardDataTree;
  84     private DataTreeModification rootModification = null;
  85
  86     private ArrayList<DOMStoreThreePhaseCommitCohort> cohorts = new ArrayList<>();
  87     private InMemoryDOMDataTreeShardChangePublisher changePublisher;
  88
  89     // FIXME inject into shard?
  90     private ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
  91
  92     InmemoryDOMDataTreeShardWriteTransaction(final ShardDataModification root,
  93                                              final DataTree rootShardDataTree,
  94                                              final InMemoryDOMDataTreeShardChangePublisher changePublisher) {
  95         this.modification = Preconditions.checkNotNull(root);
  96         this.rootShardDataTree = Preconditions.checkNotNull(rootShardDataTree);
  97         this.changePublisher = Preconditions.checkNotNull(changePublisher);
  98     }
  99
 100     private DOMDataTreeWriteCursor getCursor() {
 101         if (cursor == null) {
 102             cursor = new ShardDataModificationCursor(modification);
 103         }
 104         return cursor;
 105     }
 106
 107     void delete(final YangInstanceIdentifier path) {
 108         YangInstanceIdentifier relativePath = toRelative(path);
 109         Preconditions.checkArgument(!YangInstanceIdentifier.EMPTY.equals(relativePath),
 110                 "Deletion of shard root is not allowed");
 111         SimpleCursorOperation.DELETE.apply(getCursor(), relativePath , null);
 112     }
 113
 114     void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 115         SimpleCursorOperation.MERGE.apply(getCursor(), toRelative(path), data);
 116     }
 117
 118     void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 119         SimpleCursorOperation.DELETE.apply(getCursor(), toRelative(path), data);
 120     }
 121
 122     private YangInstanceIdentifier toRelative(final YangInstanceIdentifier path) {
 123         Optional<YangInstanceIdentifier> relative =
 124                 path.relativeTo(modification.getPrefix().getRootIdentifier());
 125         Preconditions.checkArgument(relative.isPresent());
 126         return relative.get();
 127     }
 128
 129     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
 130         // FIXME: Implement this
 131         return null;
 132     }
 133
 134     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
 135         // TODO Auto-generated method stub
 136         return null;
 137     }
 138
 139
 140     public Object getIdentifier() {
 141         // TODO Auto-generated method stub
 142         return null;
 143     }
 144
 145     public void close() {
 146         // TODO Auto-generated method stub
 147     }
 148
 149     @Override
 150     public void ready() {
 151
 152         LOG.debug("Readying open transaction on shard {}", modification.getPrefix());
 153         rootModification = modification.seal();
 154
 155         cohorts.add(new InMemoryDOMDataTreeShardThreePhaseCommitCohort(rootShardDataTree, rootModification, changePublisher));
 156         for (Entry<DOMDataTreeIdentifier, ForeignShardModificationContext> entry : modification.getChildShards().entrySet()) {
 157             cohorts.add(new ForeignShardThreePhaseCommitCohort(entry.getKey(), entry.getValue()));
 158         }
 159     }
 160
 161     @Override
 162     public ListenableFuture<Void> submit() {
 163         LOG.debug("Submitting open transaction on shard {}", modification.getPrefix());
 164
 165         Preconditions.checkNotNull(cohorts);
 166         Preconditions.checkState(!cohorts.isEmpty(), "Submitting an empty transaction");
 167
 168         final ListenableFuture<Void> submit = executor.submit(new ShardSubmitCoordinationTask(modification.getPrefix(), cohorts));
 169
 170         return submit;
 171     }
 172
 173     @Override
 174     public ListenableFuture<Boolean> validate() {
 175         LOG.debug("CanCommit on open transaction on shard {}", modification.getPrefix());
 176
 177         final ListenableFuture<Boolean> submit = executor.submit(new ShardCanCommitCoordinationTask(modification.getPrefix(), cohorts));
 178         return submit;
 179     }
 180
 181     @Override
 182     public ListenableFuture<Void> prepare() {
 183         LOG.debug("PreCommit on open transaction on shard {}", modification.getPrefix());
 184
 185         final ListenableFuture<Void> submit = executor.submit(new ShardPreCommitCoordinationTask(modification.getPrefix(), cohorts));
 186         return submit;
 187     }
 188
 189     @Override
 190     public ListenableFuture<Void> commit() {
 191         LOG.debug("Commit open transaction on shard {}", modification.getPrefix());
 192
 193         final ListenableFuture<Void> submit = executor.submit(new ShardCommitCoordinationTask(modification.getPrefix(), cohorts));
 194         return submit;
 195     }
 196
 197     public void followUp() {
 198
 199     }
 200
 201     @Override
 202     public DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
 203         DOMDataTreeWriteCursor ret = getCursor();
 204         YangInstanceIdentifier relativePath = toRelative(prefix.getRootIdentifier());
 205         ret.enter(relativePath.getPathArguments());
 206         return ret;
 207     }
 208 }