2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.remote.rpc;
12 import akka.actor.ActorRef;
13 import com.google.common.base.Preconditions;
14 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
15 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
16 import org.opendaylight.controller.remote.rpc.messages.AddRoutedRpc;
17 import org.opendaylight.controller.remote.rpc.messages.RemoveRoutedRpc;
18 import org.opendaylight.controller.remote.rpc.utils.ActorUtil;
19 import org.opendaylight.controller.sal.connector.api.RpcRouter;
20 import org.opendaylight.controller.sal.core.api.RpcRoutingContext;
21 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
25 import java.util.HashSet;
29 public class RoutedRpcListener implements RouteChangeListener<RpcRoutingContext, YangInstanceIdentifier>{
30 private static final Logger LOG = LoggerFactory.getLogger(RoutedRpcListener.class);
31 private final ActorRef rpcRegistry;
32 private final String actorPath;
34 public RoutedRpcListener(ActorRef rpcRegistry, String actorPath) {
35 Preconditions.checkNotNull(rpcRegistry, "rpc registry actor should not be null");
36 Preconditions.checkNotNull(actorPath, "actor path of rpc broker on current node should not be null");
38 this.rpcRegistry = rpcRegistry;
39 this.actorPath = actorPath;
43 public void onRouteChange(RouteChange<RpcRoutingContext, YangInstanceIdentifier> routeChange) {
44 Map<RpcRoutingContext, Set<YangInstanceIdentifier>> announcements = routeChange.getAnnouncements();
45 announce(getRouteIdentifiers(announcements));
47 Map<RpcRoutingContext, Set<YangInstanceIdentifier>> removals = routeChange.getRemovals();
48 remove(getRouteIdentifiers(removals));
53 * @param announcements
55 private void announce(Set<RpcRouter.RouteIdentifier<?, ?, ?>> announcements) {
56 LOG.debug("Announcing [{}]", announcements);
57 AddRoutedRpc addRpcMsg = new AddRoutedRpc(announcements, actorPath);
59 ActorUtil.executeLocalOperation(rpcRegistry, addRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
60 } catch (Exception e) {
61 // Just logging it because Akka API throws this exception
62 LOG.error(e.toString());
70 private void remove(Set<RpcRouter.RouteIdentifier<?, ?, ?>> removals){
71 LOG.debug("Removing [{}]", removals);
72 RemoveRoutedRpc removeRpcMsg = new RemoveRoutedRpc(removals, actorPath);
74 ActorUtil.executeLocalOperation(rpcRegistry, removeRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
75 } catch (Exception e) {
76 // Just logging it because Akka API throws this exception
77 LOG.error(e.toString());
86 private Set<RpcRouter.RouteIdentifier<?, ?, ?>> getRouteIdentifiers(Map<RpcRoutingContext, Set<YangInstanceIdentifier>> changes) {
87 RouteIdentifierImpl routeId = null;
88 Set<RpcRouter.RouteIdentifier<?, ?, ?>> routeIdSet = new HashSet<>();
90 for (RpcRoutingContext context : changes.keySet()){
91 for (YangInstanceIdentifier instanceId : changes.get(context)){
92 routeId = new RouteIdentifierImpl(null, context.getRpc(), instanceId);
93 routeIdSet.add(routeId);