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

   1 /*
   2  * Copyright (c) 2014 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 package org.opendaylight.controller.remote.rpc;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.actor.ActorRef;
  13 import akka.actor.Props;
  14 import akka.actor.Status.Failure;
  15 import com.google.common.base.Throwables;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.ListenableFuture;
  19 import com.google.common.util.concurrent.MoreExecutors;
  20 import java.util.Collection;
  21 import org.eclipse.jdt.annotation.NonNull;
  22 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor;
  23 import org.opendaylight.controller.remote.rpc.messages.ActionResponse;
  24 import org.opendaylight.controller.remote.rpc.messages.ExecuteAction;
  25 import org.opendaylight.controller.remote.rpc.messages.ExecuteRpc;
  26 import org.opendaylight.controller.remote.rpc.messages.RpcResponse;
  27 import org.opendaylight.mdsal.dom.api.DOMActionResult;
  28 import org.opendaylight.mdsal.dom.api.DOMActionService;
  29 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  30 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  31 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  32 import org.opendaylight.yangtools.yang.common.QName;
  33 import org.opendaylight.yangtools.yang.common.RpcError;
  34 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  36 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
  37
  38 /**
  39  * Actor receiving invocation requests from remote nodes, routing them to
  40  * {@link DOMRpcService#invokeRpc(SchemaPath, NormalizedNode)} and
  41  * {@link DOMActionService#invokeAction(SchemaPath, DOMDataTreeIdentifier, ContainerNode)}.
  42  *
  43  * <p>
  44  * Note that while the two interfaces are very similar, invocation strategies are slightly different due to historic
  45  * behavior of RPCs:
  46  * <ul>
  47  *   <li>RPCs allow both null input and output, and this is passed to the infrastructure. Furthermore any invocation
  48  *       which results in errors being reported drops the output content, even if it is present -- which is wrong, as
  49  *       'errors' in this case can also be just warnings.</li>
  50  *   <li>Actions do not allow null input, but allow null output. If the output is present, it is passed along with any
  51  *       errors reported.</li>
  52  * </ul>
  53  */
  54 final class OpsInvoker extends AbstractUntypedActor {
  55     private final DOMRpcService rpcService;
  56     private final DOMActionService actionService;
  57
  58     private OpsInvoker(final DOMRpcService rpcService, final DOMActionService actionService) {
  59         this.rpcService = requireNonNull(rpcService);
  60         this.actionService = requireNonNull(actionService);
  61     }
  62
  63     public static Props props(final DOMRpcService rpcService, final DOMActionService actionService) {
  64         return Props.create(OpsInvoker.class,
  65             requireNonNull(rpcService, "DOMRpcService can not be null"),
  66             requireNonNull(actionService, "DOMActionService can not be null"));
  67     }
  68
  69     @Override
  70     protected void handleReceive(final Object message) {
  71         if (message instanceof ExecuteRpc) {
  72             LOG.debug("Handling ExecuteOps Message");
  73             execute((ExecuteRpc) message);
  74         } else if (message instanceof ExecuteAction) {
  75             execute((ExecuteAction) message);
  76         } else {
  77             unknownMessage(message);
  78         }
  79     }
  80
  81     @SuppressWarnings("checkstyle:IllegalCatch")
  82     private void execute(final ExecuteRpc msg) {
  83         LOG.debug("Executing RPC {}", msg.getType());
  84         final ActorRef sender = getSender();
  85
  86         final ListenableFuture<? extends DOMRpcResult> future;
  87         try {
  88             future = rpcService.invokeRpc(msg.getType(), msg.getInput());
  89         } catch (final RuntimeException e) {
  90             LOG.debug("Failed to invoke RPC {}", msg.getType(), e);
  91             sender.tell(new Failure(e), self());
  92             return;
  93         }
  94
  95         Futures.addCallback(future, new AbstractCallback<QName, DOMRpcResult>(getSender(), msg.getType()) {
  96             @Override
  97             Object nullResponse(final QName type) {
  98                 LOG.warn("Execution of {} resulted in null result", type);
  99                 return new RpcResponse(null);
 100             }
 101
 102             @Override
 103             Object response(final QName type, final DOMRpcResult result) {
 104                 final Collection<? extends RpcError> errors = result.errors();
 105                 return errors.isEmpty() ? new RpcResponse(result.value())
 106                         // This is legacy (wrong) behavior, which ignores the fact that errors may be just warnings,
 107                         // discarding any output
 108                         : new Failure(new RpcErrorsException(String.format("Execution of rpc %s failed", type),
 109                             errors));
 110             }
 111         }, MoreExecutors.directExecutor());
 112     }
 113
 114     @SuppressWarnings("checkstyle:IllegalCatch")
 115     private void execute(final ExecuteAction msg) {
 116         LOG.debug("Executing Action {}", msg.getType());
 117
 118         final ActorRef sender = getSender();
 119
 120         final ListenableFuture<? extends DOMActionResult> future;
 121         try {
 122             future = actionService.invokeAction(msg.getType(), msg.getPath(), msg.getInput());
 123         } catch (final RuntimeException e) {
 124             LOG.debug("Failed to invoke action {}", msg.getType(), e);
 125             sender.tell(new Failure(e), self());
 126             return;
 127         }
 128
 129         Futures.addCallback(future, new AbstractCallback<Absolute, DOMActionResult>(getSender(), msg.getType()) {
 130             @Override
 131             Object nullResponse(final Absolute type) {
 132                 throw new IllegalStateException("Null invocation result of action " + type);
 133             }
 134
 135             @Override
 136             Object response(final Absolute type, final DOMActionResult result) {
 137                 final Collection<? extends RpcError> errors = result.getErrors();
 138                 return errors.isEmpty() ? new ActionResponse(result.getOutput(), result.getErrors())
 139                     // This is legacy (wrong) behavior, which ignores the fact that errors may be just warnings,
 140                     // discarding any output
 141                     : new Failure(new RpcErrorsException(String.format("Execution of action %s failed", type),
 142                         errors));
 143             }
 144         }, MoreExecutors.directExecutor());
 145     }
 146
 147     private abstract class AbstractCallback<T, R> implements FutureCallback<R> {
 148         private final ActorRef replyTo;
 149         private final T type;
 150
 151         AbstractCallback(final ActorRef replyTo, final T type) {
 152             this.replyTo = requireNonNull(replyTo);
 153             this.type = requireNonNull(type);
 154         }
 155
 156         @Override
 157         public final void onSuccess(final R result) {
 158             final Object response;
 159             if (result == null) {
 160                 // This shouldn't happen but the FutureCallback annotates the result param with Nullable so handle null
 161                 // here to avoid FindBugs warning.
 162                 response = nullResponse(type);
 163             } else {
 164                 response = response(type, result);
 165             }
 166
 167             LOG.debug("Sending response for execution of {} : {}", type, response);
 168             replyTo.tell(response, self());
 169         }
 170
 171         @Override
 172         public final void onFailure(final Throwable failure) {
 173             LOG.debug("Failed to execute operation {}", type, failure);
 174             LOG.error("Failed to execute operation {} due to {}. More details are available on DEBUG level.", type,
 175                 Throwables.getRootCause(failure).getMessage());
 176             replyTo.tell(new Failure(failure), self());
 177         }
 178
 179         abstract @NonNull Object nullResponse(@NonNull T type);
 180
 181         abstract @NonNull Object response(@NonNull T type, @NonNull R result);
 182     }
 183 }