opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/LocalTransactionContext.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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.cluster.datastore;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.actor.ActorSelection;
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import com.google.common.util.concurrent.SettableFuture;
  17 import java.util.Optional;
  18 import java.util.SortedSet;
  19 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  20 import org.opendaylight.controller.cluster.datastore.messages.AbstractRead;
  21 import org.opendaylight.mdsal.common.api.ReadFailedException;
  22 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
  23 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransaction;
  24 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
  25 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  26 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  27 import scala.concurrent.Future;
  28
  29 /**
  30  * Processes front-end transaction operations locally before being committed to the destination shard.
  31  * Instances of this class are used when the destination shard is local to the caller.
  32  *
  33  * @author Thomas Pantelis
  34  */
  35 abstract class LocalTransactionContext extends AbstractTransactionContext {
  36     private final DOMStoreTransaction txDelegate;
  37     private final LocalTransactionReadySupport readySupport;
  38     private Exception operationError;
  39
  40     LocalTransactionContext(final DOMStoreTransaction txDelegate, final TransactionIdentifier identifier,
  41             final LocalTransactionReadySupport readySupport) {
  42         super(identifier);
  43         this.txDelegate = requireNonNull(txDelegate);
  44         this.readySupport = readySupport;
  45     }
  46
  47     protected abstract DOMStoreWriteTransaction getWriteDelegate();
  48
  49     protected abstract DOMStoreReadTransaction getReadDelegate();
  50
  51     @Override
  52     @SuppressWarnings("checkstyle:IllegalCatch")
  53     public void executeDelete(final YangInstanceIdentifier path, final Boolean havePermit) {
  54         incrementModificationCount();
  55         if (operationError == null) {
  56             try {
  57                 getWriteDelegate().delete(path);
  58             } catch (Exception e) {
  59                 operationError = e;
  60             }
  61         }
  62     }
  63
  64     @Override
  65     @SuppressWarnings("checkstyle:IllegalCatch")
  66     public void executeMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
  67             final Boolean havePermit) {
  68         incrementModificationCount();
  69         if (operationError == null) {
  70             try {
  71                 getWriteDelegate().merge(path, data);
  72             } catch (Exception e) {
  73                 operationError = e;
  74             }
  75         }
  76     }
  77
  78     @Override
  79     @SuppressWarnings("checkstyle:IllegalCatch")
  80     public void executeWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
  81             final Boolean havePermit) {
  82         incrementModificationCount();
  83         if (operationError == null) {
  84             try {
  85                 getWriteDelegate().write(path, data);
  86             } catch (Exception e) {
  87                 operationError = e;
  88             }
  89         }
  90     }
  91
  92     @Override
  93     public <T> void executeRead(final AbstractRead<T> readCmd, final SettableFuture<T> proxyFuture,
  94             final Boolean havePermit) {
  95         Futures.addCallback(readCmd.apply(getReadDelegate()), new FutureCallback<T>() {
  96             @Override
  97             public void onSuccess(final T result) {
  98                 proxyFuture.set(result);
  99             }
 100
 101             @Override
 102             public void onFailure(final Throwable failure) {
 103                 proxyFuture.setException(failure instanceof Exception
 104                         ? ReadFailedException.MAPPER.apply((Exception) failure) : failure);
 105             }
 106         }, MoreExecutors.directExecutor());
 107     }
 108
 109     private LocalThreePhaseCommitCohort ready() {
 110         logModificationCount();
 111         return readySupport.onTransactionReady(getWriteDelegate(), operationError);
 112     }
 113
 114     @Override
 115     public Future<ActorSelection> readyTransaction(final Boolean havePermit,
 116             final Optional<SortedSet<String>> participatingShardNames) {
 117         final LocalThreePhaseCommitCohort cohort = ready();
 118         return cohort.initiateCoordinatedCommit(participatingShardNames);
 119     }
 120
 121     @Override
 122     public Future<Object> directCommit(final Boolean havePermit) {
 123         final LocalThreePhaseCommitCohort cohort = ready();
 124         return cohort.initiateDirectCommit();
 125     }
 126
 127     @Override
 128     public void closeTransaction() {
 129         txDelegate.close();
 130     }
 131 }