opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/AbstractRemoteFuture.java

   1 /*
   2  * Copyright (c) 2019 PANTHEON.tech, s.r.o. 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.remote.rpc;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.dispatch.OnComplete;
  13 import com.google.common.util.concurrent.AbstractFuture;
  14 import java.util.concurrent.ExecutionException;
  15 import java.util.concurrent.TimeUnit;
  16 import java.util.concurrent.TimeoutException;
  17 import org.eclipse.jdt.annotation.NonNull;
  18 import org.eclipse.jdt.annotation.Nullable;
  19 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  20 import org.slf4j.Logger;
  21 import org.slf4j.LoggerFactory;
  22 import scala.concurrent.ExecutionContext;
  23 import scala.concurrent.Future;
  24
  25 abstract class AbstractRemoteFuture<T, E extends Exception> extends AbstractFuture<T> {
  26     private static final Logger LOG = LoggerFactory.getLogger(AbstractRemoteFuture.class);
  27
  28     private final @NonNull SchemaPath type;
  29
  30     AbstractRemoteFuture(final @NonNull SchemaPath type, final Future<Object> requestFuture) {
  31         this.type = requireNonNull(type);
  32         requestFuture.onComplete(new FutureUpdater(), ExecutionContext.Implicits$.MODULE$.global());
  33     }
  34
  35     @Override
  36     public final T get() throws InterruptedException, ExecutionException {
  37         try {
  38             return super.get();
  39         } catch (ExecutionException e) {
  40             throw mapException(e);
  41         }
  42     }
  43
  44     @Override
  45     public final T get(final long timeout, final TimeUnit unit)
  46             throws InterruptedException, ExecutionException, TimeoutException {
  47         try {
  48             return super.get(timeout, unit);
  49         } catch (final ExecutionException e) {
  50             throw mapException(e);
  51         }
  52     }
  53
  54     @Override
  55     protected final boolean set(final T value) {
  56         final boolean ret = super.set(value);
  57         if (ret) {
  58             LOG.debug("Future {} for action {} successfully completed", this, type);
  59         }
  60         return ret;
  61     }
  62
  63     final void failNow(final Throwable error) {
  64         LOG.debug("Failing future {} for operation {}", this, type, error);
  65         setException(error);
  66     }
  67
  68     abstract @Nullable T processReply(Object reply);
  69
  70     abstract @NonNull Class<E> exceptionClass();
  71
  72     abstract @NonNull E wrapCause(Throwable cause);
  73
  74     private ExecutionException mapException(final ExecutionException ex) {
  75         final Throwable cause = ex.getCause();
  76         return exceptionClass().isInstance(cause) ? ex : new ExecutionException(ex.getMessage(), wrapCause(cause));
  77     }
  78
  79     private final class FutureUpdater extends OnComplete<Object> {
  80         @Override
  81         public void onComplete(final Throwable error, final Object reply) {
  82             if (error == null) {
  83                 final T result = processReply(reply);
  84                 if (result != null) {
  85                     LOG.debug("Received response for operation {}: result is {}", type, result);
  86                     set(result);
  87                 } else {
  88                     failNow(new IllegalStateException("Incorrect reply type " + reply + " from Akka"));
  89                 }
  90             } else {
  91                 failNow(error);
  92             }
  93         }
  94     }
  95 }