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