opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistry.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.registry;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.Cancellable;
  12 import akka.actor.Props;
  13 import akka.japi.Creator;
  14 import akka.japi.Option;
  15 import akka.japi.Pair;
  16 import com.google.common.base.Preconditions;
  17 import java.util.ArrayList;
  18 import java.util.Collections;
  19 import java.util.HashSet;
  20 import java.util.List;
  21 import java.util.Set;
  22 import java.util.concurrent.atomic.AtomicReference;
  23 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
  24 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
  25 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRouters;
  26 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes;
  27 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
  28 import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket;
  29 import org.opendaylight.controller.remote.rpc.registry.gossip.BucketStore;
  30 import org.opendaylight.controller.remote.rpc.registry.mbeans.RemoteRpcRegistryMXBean;
  31 import org.opendaylight.controller.remote.rpc.registry.mbeans.RemoteRpcRegistryMXBeanImpl;
  32 import org.opendaylight.controller.sal.connector.api.RpcRouter;
  33 import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier;
  34 import scala.concurrent.duration.FiniteDuration;
  35
  36 /**
  37  * Registry to look up cluster nodes that have registered for a given rpc.
  38  * <p/>
  39  * It uses {@link org.opendaylight.controller.remote.rpc.registry.gossip.BucketStore} to maintain this
  40  * cluster wide information.
  41  */
  42 public class RpcRegistry extends BucketStore<RoutingTable> {
  43     private final Set<Runnable> routesUpdatedCallbacks = new HashSet<>();
  44     private final FiniteDuration findRouterTimeout;
  45
  46     public RpcRegistry(RemoteRpcProviderConfig config) {
  47         super(config);
  48         getLocalBucket().setData(new RoutingTable());
  49         findRouterTimeout = getConfig().getGossipTickInterval().$times(10);
  50     }
  51
  52     public static Props props(RemoteRpcProviderConfig config) {
  53         return Props.create(new RpcRegistryCreator(config));
  54     }
  55
  56     @Override
  57     protected void handleReceive(Object message) throws Exception {
  58         //TODO: if sender is remote, reject message
  59
  60         if (message instanceof SetLocalRouter) {
  61             receiveSetLocalRouter((SetLocalRouter) message);
  62         } else if (message instanceof AddOrUpdateRoutes) {
  63             receiveAddRoutes((AddOrUpdateRoutes) message);
  64         } else if (message instanceof RemoveRoutes) {
  65             receiveRemoveRoutes((RemoveRoutes) message);
  66         } else if (message instanceof Messages.FindRouters) {
  67             receiveGetRouter((FindRouters) message);
  68         } else if (message instanceof Runnable) {
  69             ((Runnable)message).run();
  70         } else {
  71             super.handleReceive(message);
  72         }
  73     }
  74
  75     /**
  76      * Register's rpc broker
  77      *
  78      * @param message contains {@link akka.actor.ActorRef} for rpc broker
  79      */
  80     private void receiveSetLocalRouter(SetLocalRouter message) {
  81         getLocalBucket().getData().setRouter(message.getRouter());
  82     }
  83
  84     /**
  85      * @param msg
  86      */
  87     private void receiveAddRoutes(AddOrUpdateRoutes msg) {
  88
  89         log.debug("AddOrUpdateRoutes: {}", msg.getRouteIdentifiers());
  90
  91         RoutingTable table = getLocalBucket().getData().copy();
  92         for(RpcRouter.RouteIdentifier<?, ?, ?> routeId : msg.getRouteIdentifiers()) {
  93             table.addRoute(routeId);
  94         }
  95
  96         updateLocalBucket(table);
  97
  98         onBucketsUpdated();
  99     }
 100
 101     /**
 102      * @param msg contains list of route ids to remove
 103      */
 104     private void receiveRemoveRoutes(RemoveRoutes msg) {
 105
 106         RoutingTable table = getLocalBucket().getData().copy();
 107         for (RpcRouter.RouteIdentifier<?, ?, ?> routeId : msg.getRouteIdentifiers()) {
 108             table.removeRoute(routeId);
 109         }
 110
 111         updateLocalBucket(table);
 112     }
 113
 114     /**
 115      * Finds routers for the given rpc.
 116      *
 117      * @param findRouters
 118      */
 119     private void receiveGetRouter(final FindRouters findRouters) {
 120         log.debug("receiveGetRouter for {}", findRouters.getRouteIdentifier());
 121
 122         final ActorRef sender = getSender();
 123         if(!findRouters(findRouters, sender)) {
 124             log.debug("No routers found for {} - scheduling {} ms timer", findRouters.getRouteIdentifier(),
 125                     findRouterTimeout.toMillis());
 126
 127             final AtomicReference<Cancellable> timer = new AtomicReference<>();
 128             final Runnable routesUpdatedRunnable = new Runnable() {
 129                 @Override
 130                 public void run() {
 131                     if(findRouters(findRouters, sender)) {
 132                         routesUpdatedCallbacks.remove(this);
 133                         timer.get().cancel();
 134                     }
 135                 }
 136             };
 137
 138             routesUpdatedCallbacks.add(routesUpdatedRunnable);
 139
 140             Runnable timerRunnable = new Runnable() {
 141                 @Override
 142                 public void run() {
 143                     log.warn("Timed out finding routers for {}", findRouters.getRouteIdentifier());
 144
 145                     routesUpdatedCallbacks.remove(routesUpdatedRunnable);
 146                     sender.tell(new Messages.FindRoutersReply(
 147                             Collections.<Pair<ActorRef, Long>>emptyList()), self());
 148                 }
 149             };
 150
 151             timer.set(getContext().system().scheduler().scheduleOnce(findRouterTimeout, self(), timerRunnable,
 152                     getContext().dispatcher(), self()));
 153         }
 154     }
 155
 156     private boolean findRouters(FindRouters findRouters, ActorRef sender) {
 157         List<Pair<ActorRef, Long>> routers = new ArrayList<>();
 158
 159         RouteIdentifier<?, ?, ?> routeId = findRouters.getRouteIdentifier();
 160         findRoutes(getLocalBucket().getData(), routeId, routers);
 161
 162         for(Bucket<RoutingTable> bucket : getRemoteBuckets().values()) {
 163             findRoutes(bucket.getData(), routeId, routers);
 164         }
 165
 166         log.debug("Found {} routers for {}", routers.size(), findRouters.getRouteIdentifier());
 167
 168         boolean foundRouters = !routers.isEmpty();
 169         if(foundRouters) {
 170             sender.tell(new Messages.FindRoutersReply(routers), getSelf());
 171         }
 172
 173         return foundRouters;
 174     }
 175
 176     private void findRoutes(RoutingTable table, RpcRouter.RouteIdentifier<?, ?, ?> routeId,
 177             List<Pair<ActorRef, Long>> routers) {
 178         if (table == null) {
 179             return;
 180         }
 181
 182         Option<Pair<ActorRef, Long>> routerWithUpdateTime = table.getRouterFor(routeId);
 183         if(!routerWithUpdateTime.isEmpty()) {
 184             routers.add(routerWithUpdateTime.get());
 185         }
 186     }
 187
 188     @Override
 189     protected void onBucketsUpdated() {
 190         for(Runnable callBack: routesUpdatedCallbacks) {
 191             callBack.run();
 192         }
 193     }
 194
 195     /**
 196      * All messages used by the RpcRegistry
 197      */
 198     public static class Messages {
 199
 200
 201         public static class ContainsRoute {
 202             final List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIdentifiers;
 203
 204             public ContainsRoute(List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIdentifiers) {
 205                 Preconditions.checkArgument(routeIdentifiers != null &&
 206                                             !routeIdentifiers.isEmpty(),
 207                                             "Route Identifiers must be supplied");
 208                 this.routeIdentifiers = routeIdentifiers;
 209             }
 210
 211             public List<RpcRouter.RouteIdentifier<?, ?, ?>> getRouteIdentifiers() {
 212                 return this.routeIdentifiers;
 213             }
 214
 215             @Override
 216             public String toString() {
 217                 return "ContainsRoute{" +
 218                         "routeIdentifiers=" + routeIdentifiers +
 219                         '}';
 220             }
 221         }
 222
 223         public static class AddOrUpdateRoutes extends ContainsRoute {
 224
 225             public AddOrUpdateRoutes(List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIdentifiers) {
 226                 super(routeIdentifiers);
 227             }
 228         }
 229
 230         public static class RemoveRoutes extends ContainsRoute {
 231
 232             public RemoveRoutes(List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIdentifiers) {
 233                 super(routeIdentifiers);
 234             }
 235         }
 236
 237         public static class SetLocalRouter {
 238             private final ActorRef router;
 239
 240             public SetLocalRouter(ActorRef router) {
 241                 Preconditions.checkArgument(router != null, "Router must not be null");
 242                 this.router = router;
 243             }
 244
 245             public ActorRef getRouter() {
 246                 return this.router;
 247             }
 248
 249             @Override
 250             public String toString() {
 251                 return "SetLocalRouter{" +
 252                         "router=" + router +
 253                         '}';
 254             }
 255         }
 256
 257         public static class FindRouters {
 258             private final RpcRouter.RouteIdentifier<?, ?, ?> routeIdentifier;
 259
 260             public FindRouters(RpcRouter.RouteIdentifier<?, ?, ?> routeIdentifier) {
 261                 Preconditions.checkArgument(routeIdentifier != null, "Route must not be null");
 262                 this.routeIdentifier = routeIdentifier;
 263             }
 264
 265             public RpcRouter.RouteIdentifier<?, ?, ?> getRouteIdentifier() {
 266                 return routeIdentifier;
 267             }
 268
 269             @Override
 270             public String toString() {
 271                 return "FindRouters{" +
 272                         "routeIdentifier=" + routeIdentifier +
 273                         '}';
 274             }
 275         }
 276
 277         public static class FindRoutersReply {
 278             final List<Pair<ActorRef, Long>> routerWithUpdateTime;
 279
 280             public FindRoutersReply(List<Pair<ActorRef, Long>> routerWithUpdateTime) {
 281                 Preconditions.checkArgument(routerWithUpdateTime != null, "List of routers found must not be null");
 282                 this.routerWithUpdateTime = routerWithUpdateTime;
 283             }
 284
 285             public List<Pair<ActorRef, Long>> getRouterWithUpdateTime() {
 286                 return routerWithUpdateTime;
 287             }
 288
 289             @Override
 290             public String toString() {
 291                 return "FindRoutersReply{" +
 292                         "routerWithUpdateTime=" + routerWithUpdateTime +
 293                         '}';
 294             }
 295         }
 296     }
 297
 298     private static class RpcRegistryCreator implements Creator<RpcRegistry> {
 299         private static final long serialVersionUID = 1L;
 300         private final RemoteRpcProviderConfig config;
 301
 302         private RpcRegistryCreator(RemoteRpcProviderConfig config) {
 303             this.config = config;
 304         }
 305
 306         @Override
 307         public RpcRegistry create() throws Exception {
 308             RpcRegistry registry =  new RpcRegistry(config);
 309             RemoteRpcRegistryMXBean mxBean = new RemoteRpcRegistryMXBeanImpl(registry);
 310             return registry;
 311         }
 312     }
 313 }