binding/mdsal-binding-util/src/main/java/org/opendaylight/mdsal/binding/util/RetryingManagedNewTransactionRunnerImpl.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.mdsal.binding.util;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.util.concurrent.FluentFuture;
  13 import com.google.common.util.concurrent.MoreExecutors;
  14 import java.util.concurrent.ExecutionException;
  15 import java.util.concurrent.Executor;
  16 import java.util.function.Function;
  17 import org.opendaylight.mdsal.common.api.OptimisticLockFailedException;
  18 import org.opendaylight.mdsal.common.api.ReadFailedException;
  19
  20 /**
  21  * Implementation of {@link ManagedNewTransactionRunner} with automatic transparent retries on transaction failure
  22  * ({@link OptimisticLockFailedException} on write transactions and {@link ReadFailedException} on read transactions
  23  * will cause the operation constructing the transaction to be re-run).
  24  * This is a package local private internal class; end-users use the {@link RetryingManagedNewTransactionRunner}.
  25  * @see RetryingManagedNewTransactionRunner
  26  *
  27  * @author Michael Vorburger.ch &amp; Stephen Kitt, with input from Tom Pantelis re. catchingAsync &amp; direct Executor
  28  */
  29 // intentionally package local
  30 class RetryingManagedNewTransactionRunnerImpl implements ManagedNewTransactionRunner {
  31
  32     // NB: The RetryingManagedNewTransactionRunnerTest is in mdsalutil-testutils's src/test, not this project's
  33
  34     // duplicated in SingleTransactionDataBroker
  35     private static final int DEFAULT_RETRIES = 3;
  36
  37     private final int maxRetries;
  38
  39     private final ManagedNewTransactionRunner delegate;
  40
  41     private final Executor executor;
  42
  43     RetryingManagedNewTransactionRunnerImpl(ManagedNewTransactionRunner delegate) {
  44         this(delegate, MoreExecutors.directExecutor(), DEFAULT_RETRIES);
  45     }
  46
  47     RetryingManagedNewTransactionRunnerImpl(ManagedNewTransactionRunner delegate, int maxRetries) {
  48         this(delegate, MoreExecutors.directExecutor(), maxRetries);
  49     }
  50
  51     RetryingManagedNewTransactionRunnerImpl(ManagedNewTransactionRunner delegate, Executor executor, int maxRetries) {
  52         this.delegate = requireNonNull(delegate, "delegate must not be null");
  53         this.executor = requireNonNull(executor, "executor must not be null");
  54         this.maxRetries = maxRetries;
  55     }
  56
  57     @Override
  58     public <D extends Datastore, E extends Exception, R> R applyInterruptiblyWithNewReadOnlyTransactionAndClose(
  59             Class<D> datastoreType, InterruptibleCheckedFunction<TypedReadTransaction<D>, R, E> txFunction)
  60             throws E, InterruptedException {
  61         return applyInterruptiblyWithNewReadOnlyTransactionAndClose(datastoreType, txFunction, maxRetries);
  62     }
  63
  64     @SuppressWarnings("checkstyle:IllegalCatch")
  65     private <R, D extends Datastore, E extends Exception> R applyInterruptiblyWithNewReadOnlyTransactionAndClose(
  66             Class<D> datastoreType, InterruptibleCheckedFunction<TypedReadTransaction<D>, R, E> txFunction,
  67             int tries) throws E, InterruptedException {
  68         try {
  69             return delegate.applyInterruptiblyWithNewReadOnlyTransactionAndClose(datastoreType, txFunction);
  70         } catch (Exception e) {
  71             if (isRetriableException(e) && tries - 1 > 0) {
  72                 return applyInterruptiblyWithNewReadOnlyTransactionAndClose(datastoreType, txFunction, tries - 1);
  73             } else {
  74                 throw e;
  75             }
  76         }
  77     }
  78
  79     @Override
  80     public <D extends Datastore, E extends Exception, R> R applyWithNewReadOnlyTransactionAndClose(
  81             Class<D> datastoreType, CheckedFunction<TypedReadTransaction<D>, R, E> txFunction) throws E {
  82         return applyWithNewReadOnlyTransactionAndClose(datastoreType, txFunction, maxRetries);
  83     }
  84
  85     @SuppressWarnings("checkstyle:IllegalCatch")
  86     private <R, D extends Datastore, E extends Exception> R applyWithNewReadOnlyTransactionAndClose(
  87         Class<D> datastoreType, CheckedFunction<TypedReadTransaction<D>, R, E> txFunction, int tries) throws E {
  88         try {
  89             return delegate.applyWithNewReadOnlyTransactionAndClose(datastoreType, txFunction);
  90         } catch (Exception e) {
  91             if (isRetriableException(e) && tries - 1 > 0) {
  92                 return applyWithNewReadOnlyTransactionAndClose(datastoreType, txFunction, tries - 1);
  93             } else {
  94                 throw e;
  95             }
  96         }
  97     }
  98
  99     @Override
 100     public <D extends Datastore, E extends Exception, R> FluentFuture<R> applyWithNewReadWriteTransactionAndSubmit(
 101             Class<D> datastoreType, InterruptibleCheckedFunction<TypedReadWriteTransaction<D>, R, E> txFunction) {
 102         return applyWithNewReadWriteTransactionAndSubmit(datastoreType, txFunction, maxRetries);
 103     }
 104
 105     private <D extends Datastore, E extends Exception, R> FluentFuture<R> applyWithNewReadWriteTransactionAndSubmit(
 106             Class<D> datastoreType, InterruptibleCheckedFunction<TypedReadWriteTransaction<D>, R, E> txRunner,
 107             int tries) {
 108         FluentFuture<R> future = requireNonNull(
 109             delegate.applyWithNewReadWriteTransactionAndSubmit(datastoreType, txRunner),
 110             "delegate.callWithNewReadWriteTransactionAndSubmit() == null");
 111         return future.catchingAsync(Exception.class, exception -> {
 112             if (isRetriableException(exception) && tries - 1 > 0) {
 113                 return applyWithNewReadWriteTransactionAndSubmit(datastoreType, txRunner, tries - 1);
 114             } else {
 115                 throw exception;
 116             }
 117         }, executor);
 118     }
 119
 120     @Override
 121     public <R> R applyWithNewTransactionChainAndClose(Function<ManagedTransactionChain, R> chainConsumer) {
 122         throw new UnsupportedOperationException("The retrying transaction manager doesn't support transaction chains");
 123     }
 124
 125     @Override
 126     public <D extends Datastore, E extends Exception> void callInterruptiblyWithNewReadOnlyTransactionAndClose(
 127             Class<D> datastoreType, InterruptibleCheckedConsumer<TypedReadTransaction<D>, E> txConsumer)
 128             throws E, InterruptedException {
 129         callInterruptiblyWithNewReadOnlyTransactionAndClose(datastoreType, txConsumer, maxRetries);
 130     }
 131
 132     @SuppressWarnings("checkstyle:IllegalCatch")
 133     private <D extends Datastore, E extends Exception> void callInterruptiblyWithNewReadOnlyTransactionAndClose(
 134             Class<D> datastoreType, InterruptibleCheckedConsumer<TypedReadTransaction<D>, E> txConsumer, int tries)
 135             throws E, InterruptedException {
 136         try {
 137             delegate.callInterruptiblyWithNewReadOnlyTransactionAndClose(datastoreType, txConsumer);
 138         } catch (Exception e) {
 139             if (isRetriableException(e) && tries - 1 > 0) {
 140                 callInterruptiblyWithNewReadOnlyTransactionAndClose(datastoreType, txConsumer, tries - 1);
 141             } else {
 142                 throw e;
 143             }
 144         }
 145     }
 146
 147     @Override
 148     public <D extends Datastore, E extends Exception> void callWithNewReadOnlyTransactionAndClose(
 149             Class<D> datastoreType, CheckedConsumer<TypedReadTransaction<D>, E> txConsumer) throws E {
 150         callWithNewReadOnlyTransactionAndClose(datastoreType, txConsumer, maxRetries);
 151     }
 152
 153     @SuppressWarnings("checkstyle:IllegalCatch")
 154     private <D extends Datastore, E extends Exception> void callWithNewReadOnlyTransactionAndClose(
 155             Class<D> datastoreType, CheckedConsumer<TypedReadTransaction<D>, E> txConsumer, int tries) throws E {
 156         try {
 157             delegate.callWithNewReadOnlyTransactionAndClose(datastoreType, txConsumer);
 158         } catch (Exception e) {
 159             if (isRetriableException(e) && tries - 1 > 0) {
 160                 callWithNewReadOnlyTransactionAndClose(datastoreType, txConsumer, tries - 1);
 161             } else {
 162                 throw e;
 163             }
 164         }
 165     }
 166
 167     @Override
 168     public <D extends Datastore, E extends Exception> FluentFuture<? extends Object>
 169         callWithNewReadWriteTransactionAndSubmit(
 170             Class<D> datastoreType, InterruptibleCheckedConsumer<TypedReadWriteTransaction<D>, E> txConsumer) {
 171         return callWithNewReadWriteTransactionAndSubmit(datastoreType, txConsumer, maxRetries);
 172     }
 173
 174     private <D extends Datastore, E extends Exception, T> FluentFuture<T>
 175         callWithNewReadWriteTransactionAndSubmit(Class<D> datastoreType,
 176             InterruptibleCheckedConsumer<TypedReadWriteTransaction<D>, E> txRunner, int tries) {
 177
 178         return (FluentFuture<T>) requireNonNull(
 179             delegate.callWithNewReadWriteTransactionAndSubmit(datastoreType, txRunner),
 180             "delegate.callWithNewWriteOnlyTransactionAndSubmit() == null")
 181             .catchingAsync(Exception.class, exception -> {
 182                 // as per AsyncWriteTransaction.submit()'s JavaDoc re. retries
 183                 if (isRetriableException(exception) && tries - 1 > 0) {
 184                     return callWithNewReadWriteTransactionAndSubmit(datastoreType, txRunner, tries - 1);
 185                 } else {
 186                     // out of retries, so propagate the exception
 187                     throw exception;
 188                 }
 189             }, executor);
 190     }
 191
 192     @Override
 193     public <D extends Datastore, E extends Exception> FluentFuture<? extends Object>
 194         callWithNewWriteOnlyTransactionAndSubmit(Class<D> datastoreType,
 195             InterruptibleCheckedConsumer<TypedWriteTransaction<D>, E> txConsumer) {
 196         return callWithNewWriteOnlyTransactionAndSubmit(datastoreType, txConsumer, maxRetries);
 197     }
 198
 199     private <D extends Datastore, E extends Exception, T> FluentFuture<T>
 200         callWithNewWriteOnlyTransactionAndSubmit(Class<D> datastoreType,
 201             InterruptibleCheckedConsumer<TypedWriteTransaction<D>, E> txRunner, int tries) {
 202
 203         return (FluentFuture<T>) requireNonNull(
 204             delegate.callWithNewWriteOnlyTransactionAndSubmit(datastoreType, txRunner),
 205             "delegate.callWithNewWriteOnlyTransactionAndSubmit() == null")
 206             .catchingAsync(OptimisticLockFailedException.class, optimisticLockFailedException -> {
 207                 // as per AsyncWriteTransaction.submit()'s JavaDoc re. retries
 208                 if (tries - 1 > 0) {
 209                     return callWithNewWriteOnlyTransactionAndSubmit(datastoreType, txRunner, tries - 1);
 210                 } else {
 211                     // out of retries, so propagate the OptimisticLockFailedException
 212                     throw optimisticLockFailedException;
 213                 }
 214             }, executor);
 215     }
 216
 217     private boolean isRetriableException(Throwable throwable) {
 218         return throwable instanceof OptimisticLockFailedException || throwable instanceof ReadFailedException || (
 219             throwable instanceof ExecutionException && isRetriableException(throwable.getCause()));
 220     }
 221 }