Teach sal-remoterpc-connector to route actions
[controller.git] / 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 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.