opendaylight/md-sal/sal-remoterpc-connector/implementation/src/main/java/org/opendaylight/controller/sal/connector/remoterpc/RemoteRpcProvider.java

   1 /*
   2  * Copyright (c) 2013 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
   9 package org.opendaylight.controller.sal.connector.remoterpc;
  10
  11 import com.google.common.base.Optional;
  12 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
  13 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
  14 import org.opendaylight.controller.sal.connector.api.RpcRouter;
  15 import org.opendaylight.controller.sal.connector.remoterpc.api.RoutingTable;
  16 import org.opendaylight.controller.sal.connector.remoterpc.api.RoutingTableException;
  17 import org.opendaylight.controller.sal.connector.remoterpc.api.SystemException;
  18 import org.opendaylight.controller.sal.connector.remoterpc.dto.RouteIdentifierImpl;
  19 import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;
  20 import org.opendaylight.controller.sal.core.api.Provider;
  21 import org.opendaylight.controller.sal.core.api.RoutedRpcDefaultImplementation;
  22 import org.opendaylight.controller.sal.core.api.RpcImplementation;
  23 import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry;
  24 import org.opendaylight.controller.sal.core.api.RpcRegistrationListener;
  25 import org.opendaylight.controller.sal.core.api.RpcRoutingContext;
  26 import org.opendaylight.yangtools.yang.common.QName;
  27 import org.opendaylight.yangtools.yang.common.RpcResult;
  28 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
  29 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  30 import org.osgi.framework.BundleContext;
  31 import org.osgi.util.tracker.ServiceTracker;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 import java.util.Collection;
  36 import java.util.Collections;
  37 import java.util.HashSet;
  38 import java.util.Map;
  39 import java.util.Set;
  40
  41 import static com.google.common.base.Preconditions.checkNotNull;
  42
  43 public class RemoteRpcProvider implements
  44     RpcImplementation,
  45     RoutedRpcDefaultImplementation,
  46     AutoCloseable,
  47     Provider {
  48
  49   private Logger _logger = LoggerFactory.getLogger(RemoteRpcProvider.class);
  50
  51   private final ServerImpl server;
  52   private final ClientImpl client;
  53   private RoutingTableProvider routingTableProvider;
  54   private final RpcListener listener = new RpcListener();
  55   private final RoutedRpcListener routeChangeListener = new RoutedRpcListener();
  56   private ProviderSession brokerSession;
  57   private RpcProvisionRegistry rpcProvisionRegistry;
  58   private BundleContext context;
  59   private ServiceTracker clusterTracker;
  60
  61   public RemoteRpcProvider(ServerImpl server, ClientImpl client) {
  62     this.server = server;
  63     this.client = client;
  64   }
  65
  66   public void setRoutingTableProvider(RoutingTableProvider provider) {
  67     this.routingTableProvider = provider;
  68     client.setRoutingTableProvider(provider);
  69   }
  70
  71   public void setContext(BundleContext context){
  72     this.context = context;
  73   }
  74
  75   public void setRpcProvisionRegistry(RpcProvisionRegistry rpcProvisionRegistry){
  76     this.rpcProvisionRegistry = rpcProvisionRegistry;
  77   }
  78
  79   @Override
  80   public void onSessionInitiated(ProviderSession session) {
  81     brokerSession = session;
  82     server.setBrokerSession(session);
  83     start();
  84   }
  85
  86   @Override
  87   public Set<QName> getSupportedRpcs() {
  88     //TODO: Ask Tony if we need to get this from routing table
  89     return Collections.emptySet();
  90   }
  91
  92   @Override
  93   public Collection<ProviderFunctionality> getProviderFunctionality() {
  94     // TODO Auto-generated method stub
  95     return null;
  96   }
  97
  98   @Override
  99   public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
 100     return client.invokeRpc(rpc, input);
 101   }
 102
 103   @Override
 104   public RpcResult<CompositeNode> invokeRpc(QName rpc, InstanceIdentifier identifier, CompositeNode input) {
 105     return client.invokeRpc(rpc, identifier, input);
 106   }
 107
 108   public void start() {
 109     server.start();
 110     client.start();
 111     brokerSession.addRpcRegistrationListener(listener);
 112     rpcProvisionRegistry.setRoutedRpcDefaultDelegate(this);
 113     rpcProvisionRegistry.registerRouteChangeListener(routeChangeListener);
 114
 115     announceSupportedRpcs();
 116     announceSupportedRoutedRpcs();
 117   }
 118
 119   @Override
 120   public void close() throws Exception {
 121     unregisterSupportedRpcs();
 122     unregisterSupportedRoutedRpcs();
 123     server.close();
 124     client.close();
 125   }
 126
 127   public void stop() {
 128     server.stop();
 129     client.stop();
 130   }
 131
 132   /**
 133    * Add all the locally registered RPCs in the clustered routing table
 134    */
 135   private void announceSupportedRpcs(){
 136     Set<QName> currentlySupported = brokerSession.getSupportedRpcs();
 137     for (QName rpc : currentlySupported) {
 138       listener.onRpcImplementationAdded(rpc);
 139     }
 140   }
 141
 142   /**
 143    * Add all the locally registered Routed RPCs in the clustered routing table
 144    */
 145   private void announceSupportedRoutedRpcs(){
 146
 147     //TODO: announce all routed RPCs as well
 148
 149   }
 150
 151   /**
 152    * Un-Register all the supported RPCs from clustered routing table
 153    */
 154   private void unregisterSupportedRpcs(){
 155     Set<QName> currentlySupported = brokerSession.getSupportedRpcs();
 156     //TODO: remove all routed RPCs as well
 157     for (QName rpc : currentlySupported) {
 158       listener.onRpcImplementationRemoved(rpc);
 159     }
 160   }
 161
 162   /**
 163    * Un-Register all the locally supported Routed RPCs from clustered routing table
 164    */
 165   private void unregisterSupportedRoutedRpcs(){
 166
 167     //TODO: remove all routed RPCs as well
 168
 169   }
 170
 171   private RoutingTable<RpcRouter.RouteIdentifier, String> getRoutingTable(){
 172     Optional<RoutingTable<RpcRouter.RouteIdentifier, String>> routingTable =
 173         routingTableProvider.getRoutingTable();
 174
 175     checkNotNull(routingTable.isPresent(), "Routing table is null");
 176
 177     return routingTable.get();
 178   }
 179
 180   /**
 181    * Listener for rpc registrations in broker
 182    */
 183   private class RpcListener implements RpcRegistrationListener {
 184
 185     @Override
 186     public void onRpcImplementationAdded(QName rpc) {
 187
 188       _logger.debug("Adding registration for [{}]", rpc);
 189       RouteIdentifierImpl routeId = new RouteIdentifierImpl();
 190       routeId.setType(rpc);
 191
 192       RoutingTable<RpcRouter.RouteIdentifier, String> routingTable = getRoutingTable();
 193
 194       try {
 195         routingTable.addGlobalRoute(routeId, server.getServerAddress());
 196         _logger.debug("Route added [{}-{}]", routeId, server.getServerAddress());
 197
 198       } catch (RoutingTableException | SystemException e) {
 199         //TODO: This can be thrown when route already exists in the table. Broker
 200         //needs to handle this.
 201         _logger.error("Unhandled exception while adding global route to routing table [{}]", e);
 202
 203       }
 204     }
 205
 206     @Override
 207     public void onRpcImplementationRemoved(QName rpc) {
 208
 209       _logger.debug("Removing registration for [{}]", rpc);
 210       RouteIdentifierImpl routeId = new RouteIdentifierImpl();
 211       routeId.setType(rpc);
 212
 213       RoutingTable<RpcRouter.RouteIdentifier, String> routingTable = getRoutingTable();
 214
 215       try {
 216         routingTable.removeGlobalRoute(routeId);
 217       } catch (RoutingTableException | SystemException e) {
 218         _logger.error("Route delete failed {}", e);
 219       }
 220     }
 221   }
 222
 223   /**
 224    * Listener for Routed Rpc registrations in broker
 225    */
 226   private class RoutedRpcListener
 227       implements RouteChangeListener<RpcRoutingContext, InstanceIdentifier> {
 228
 229     /**
 230      *
 231      * @param routeChange
 232      */
 233     @Override
 234     public void onRouteChange(RouteChange<RpcRoutingContext, InstanceIdentifier> routeChange) {
 235       Map<RpcRoutingContext, Set<InstanceIdentifier>> announcements = routeChange.getAnnouncements();
 236       announce(getRouteIdentifiers(announcements));
 237
 238       Map<RpcRoutingContext, Set<InstanceIdentifier>> removals = routeChange.getRemovals();
 239       remove(getRouteIdentifiers(removals));
 240     }
 241
 242     /**
 243      *
 244      * @param announcements
 245      */
 246     private void announce(Set<RpcRouter.RouteIdentifier> announcements) {
 247       _logger.debug("Announcing [{}]", announcements);
 248       RoutingTable<RpcRouter.RouteIdentifier, String> routingTable = getRoutingTable();
 249       try {
 250         routingTable.addRoutes(announcements, server.getServerAddress());
 251       } catch (RoutingTableException | SystemException e) {
 252         _logger.error("Route announcement failed {}", e);
 253       }
 254     }
 255
 256     /**
 257      *
 258      * @param removals
 259      */
 260     private void remove(Set<RpcRouter.RouteIdentifier> removals){
 261       _logger.debug("Removing [{}]", removals);
 262       RoutingTable<RpcRouter.RouteIdentifier, String> routingTable = getRoutingTable();
 263       try {
 264         routingTable.removeRoutes(removals, server.getServerAddress());
 265       } catch (RoutingTableException | SystemException e) {
 266         _logger.error("Route removal failed {}", e);
 267       }
 268     }
 269
 270     /**
 271      *
 272      * @param changes
 273      * @return
 274      */
 275     private Set<RpcRouter.RouteIdentifier> getRouteIdentifiers(Map<RpcRoutingContext, Set<InstanceIdentifier>> changes) {
 276       RouteIdentifierImpl routeId = null;
 277       Set<RpcRouter.RouteIdentifier> routeIdSet = new HashSet<RpcRouter.RouteIdentifier>();
 278
 279       for (RpcRoutingContext context : changes.keySet()){
 280         routeId = new RouteIdentifierImpl();
 281         routeId.setType(context.getRpc());
 282         routeId.setContext(context.getContext());
 283
 284         for (InstanceIdentifier instanceId : changes.get(context)){
 285           routeId.setRoute(instanceId);
 286           routeIdSet.add(routeId);
 287         }
 288       }
 289       return routeIdSet;
 290     }
 291
 292
 293
 294   }
 295
 296 }