mdsalutil/mdsalutil-api/src/main/java/org/opendaylight/genius/infra/ManagedNewTransactionRunnerImpl.java

   1 /*
   2  * Copyright (c) 2017 Red Hat, 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.genius.infra;
   9
  10 import static com.google.common.util.concurrent.Futures.immediateFailedFuture;
  11
  12 import com.google.common.annotations.Beta;
  13 import com.google.common.util.concurrent.FluentFuture;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import com.google.common.util.concurrent.MoreExecutors;
  17 import java.util.function.Function;
  18 import javax.inject.Inject;
  19 import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
  20 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  21 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  22 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  23 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  24 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  25 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  26 import org.opendaylight.infrautils.utils.function.InterruptibleCheckedConsumer;
  27 import org.opendaylight.infrautils.utils.function.InterruptibleCheckedFunction;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 /**
  32  * Implementation of {@link ManagedNewTransactionRunner}. This is based on {@link ManagedTransactionFactoryImpl} but
  33  * re-implements operations based on read-write transactions to cancel transactions which don't end up making any
  34  * changes to the datastore.
  35  */
  36 @Beta
  37 // Do *NOT* mark this as @Singleton, because users choose Impl; and as long as this in API, because of https://wiki.opendaylight.org/view/BestPractices/DI_Guidelines#Nota_Bene
  38 public class ManagedNewTransactionRunnerImpl extends ManagedTransactionFactoryImpl
  39         implements ManagedNewTransactionRunner {
  40
  41     private static final Logger LOG = LoggerFactory.getLogger(ManagedNewTransactionRunnerImpl.class);
  42
  43     private final DataBroker broker;
  44
  45     @Inject
  46     public ManagedNewTransactionRunnerImpl(DataBroker broker) {
  47         super(broker);
  48         this.broker = broker;
  49     }
  50
  51     @Override
  52     @SuppressWarnings("checkstyle:IllegalCatch")
  53     public <E extends Exception> ListenableFuture<Void>
  54             callWithNewWriteOnlyTransactionAndSubmit(InterruptibleCheckedConsumer<WriteTransaction, E> txConsumer) {
  55         WriteTransaction realTx = broker.newWriteOnlyTransaction();
  56         WriteTransaction wrappedTx = new NonSubmitCancelableWriteTransaction(realTx);
  57         try {
  58             txConsumer.accept(wrappedTx);
  59             return realTx.submit();
  60         // catch Exception for both the <E extends Exception> thrown by accept() as well as any RuntimeException
  61         } catch (Exception e) {
  62             if (!realTx.cancel()) {
  63                 LOG.error("Transaction.cancel() return false - this should never happen (here)");
  64             }
  65             return immediateFailedFuture(e);
  66         }
  67     }
  68
  69     @Override
  70     @SuppressWarnings("checkstyle:IllegalCatch")
  71     public <E extends Exception> ListenableFuture<Void>
  72             callWithNewReadWriteTransactionAndSubmit(InterruptibleCheckedConsumer<ReadWriteTransaction, E> txConsumer) {
  73         ReadWriteTransaction realTx = broker.newReadWriteTransaction();
  74         WriteTrackingReadWriteTransaction wrappedTx = new NonSubmitCancelableReadWriteTransaction(realTx);
  75         try {
  76             txConsumer.accept(wrappedTx);
  77             if (wrappedTx.isWritten()) {
  78                 // The transaction contains changes, commit it
  79                 return realTx.submit();
  80             } else {
  81                 // The transaction only handled reads, cancel it
  82                 realTx.cancel();
  83                 return Futures.immediateCheckedFuture(null);
  84             }
  85         // catch Exception for both the <E extends Exception> thrown by accept() as well as any RuntimeException
  86         } catch (Exception e) {
  87             if (!realTx.cancel()) {
  88                 LOG.error("Transaction.cancel() returned false, which should never happen here");
  89             }
  90             return immediateFailedFuture(e);
  91         }
  92     }
  93
  94     @Override
  95     @SuppressWarnings("checkstyle:IllegalCatch")
  96     public <D extends Datastore, E extends Exception> FluentFuture<Void>
  97         callWithNewReadWriteTransactionAndSubmit(Class<D> datastoreType,
  98             InterruptibleCheckedConsumer<TypedReadWriteTransaction<D>, E> txConsumer) {
  99         ReadWriteTransaction realTx = broker.newReadWriteTransaction();
 100         WriteTrackingTypedReadWriteTransactionImpl<D> wrappedTx =
 101             new WriteTrackingTypedReadWriteTransactionImpl<>(datastoreType, realTx);
 102         try {
 103             txConsumer.accept(wrappedTx);
 104             if (wrappedTx.isWritten()) {
 105                 // The transaction contains changes, commit it
 106                 return realTx.commit().transform(v -> null, MoreExecutors.directExecutor());
 107             } else {
 108                 // The transaction only handled reads, cancel it
 109                 realTx.cancel();
 110                 return FluentFuture.from(Futures.immediateFuture(null));
 111             }
 112             // catch Exception for both the <E extends Exception> thrown by accept() as well as any RuntimeException
 113         } catch (Exception e) {
 114             if (!realTx.cancel()) {
 115                 LOG.error("Transaction.cancel() returned false, which should never happen here");
 116             }
 117             return FluentFuture.from(immediateFailedFuture(e));
 118         }
 119     }
 120
 121     @Override
 122     @SuppressWarnings("checkstyle:IllegalCatch")
 123     public <D extends Datastore, E extends Exception, R> FluentFuture<R> applyWithNewReadWriteTransactionAndSubmit(
 124             Class<D> datastoreType, InterruptibleCheckedFunction<TypedReadWriteTransaction<D>, R, E> txFunction) {
 125         ReadWriteTransaction realTx = broker.newReadWriteTransaction();
 126         WriteTrackingTypedReadWriteTransactionImpl<D> wrappedTx =
 127                 new WriteTrackingTypedReadWriteTransactionImpl<>(datastoreType, realTx);
 128         try {
 129             R result = txFunction.apply(wrappedTx);
 130             if (wrappedTx.isWritten()) {
 131                 // The transaction contains changes, commit it
 132                 return realTx.commit().transform(v -> result, MoreExecutors.directExecutor());
 133             } else {
 134                 // The transaction only handled reads, cancel it
 135                 realTx.cancel();
 136                 return FluentFuture.from(Futures.immediateFuture(result));
 137             }
 138             // catch Exception for both the <E extends Exception> thrown by accept() as well as any RuntimeException
 139         } catch (Exception e) {
 140             if (!realTx.cancel()) {
 141                 LOG.error("Transaction.cancel() returned false, which should never happen here");
 142             }
 143             return FluentFuture.from(immediateFailedFuture(e));
 144         }
 145     }
 146
 147     @Override
 148     public <R> R applyWithNewTransactionChainAndClose(Function<ManagedTransactionChain, R> chainConsumer) {
 149         try (BindingTransactionChain realTxChain = broker.createTransactionChain(new TransactionChainListener() {
 150             @Override
 151             public void onTransactionChainFailed(TransactionChain<?, ?> chain, AsyncTransaction<?, ?> transaction,
 152                 Throwable cause) {
 153                 LOG.error("Error handling a transaction chain", cause);
 154             }
 155
 156             @Override
 157             public void onTransactionChainSuccessful(TransactionChain<?, ?> chain) {
 158                 // Nothing to do
 159             }
 160         })) {
 161             return chainConsumer.apply(new ManagedTransactionChainImpl(realTxChain));
 162         }
 163     }
 164 }