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