opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/impl/SchemaAwareRpcBroker.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.sal.dom.broker.impl;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Preconditions.checkState;
  12
  13 import java.util.Set;
  14 import java.util.concurrent.ConcurrentHashMap;
  15 import java.util.concurrent.ConcurrentMap;
  16
  17 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
  18 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
  19 import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
  20 import org.opendaylight.controller.sal.core.api.Broker.RoutedRpcRegistration;
  21 import org.opendaylight.controller.sal.core.api.Broker.RpcRegistration;
  22 import org.opendaylight.controller.sal.core.api.RpcImplementation;
  23 import org.opendaylight.controller.sal.core.api.RpcRegistrationListener;
  24 import org.opendaylight.controller.sal.core.api.RpcRoutingContext;
  25 import org.opendaylight.controller.sal.dom.broker.spi.RpcRouter;
  26 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
  27 import org.opendaylight.yangtools.concepts.Identifiable;
  28 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  29 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
  30 import org.opendaylight.yangtools.yang.common.QName;
  31 import org.opendaylight.yangtools.yang.common.RpcResult;
  32 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
  33 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
  34 import org.opendaylight.yangtools.yang.data.api.SimpleNode;
  35 import org.opendaylight.yangtools.yang.model.api.ContainerSchemaNode;
  36 import org.opendaylight.yangtools.yang.model.api.DataSchemaNode;
  37 import org.opendaylight.yangtools.yang.model.api.Module;
  38 import org.opendaylight.yangtools.yang.model.api.RpcDefinition;
  39 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  40 import org.opendaylight.yangtools.yang.model.api.UnknownSchemaNode;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 import com.google.common.base.Optional;
  45 import com.google.common.collect.ImmutableSet;
  46
  47 public class SchemaAwareRpcBroker implements RpcRouter, Identifiable<String> {
  48
  49     private static final Logger LOG = LoggerFactory.getLogger(SchemaAwareRpcBroker.class);
  50
  51     private static final QName CONTEXT_REFERENCE = QName.create("urn:opendaylight:yang:extension:yang-ext",
  52             "2013-07-09", "context-reference");
  53     private final ListenerRegistry<RpcRegistrationListener> rpcRegistrationListeners = new ListenerRegistry<>();
  54     private final ListenerRegistry<RouteChangeListener<RpcRoutingContext, InstanceIdentifier>> routeChangeListeners = new ListenerRegistry<>();
  55
  56
  57     private final String identifier;
  58     private final ConcurrentMap<QName, RpcImplementation> implementations = new ConcurrentHashMap<>();
  59     private RpcImplementation defaultImplementation;
  60     private SchemaContextProvider schemaProvider;
  61
  62     public SchemaAwareRpcBroker(String identifier, SchemaContextProvider schemaProvider) {
  63         super();
  64         this.identifier = identifier;
  65         this.schemaProvider = schemaProvider;
  66     }
  67
  68     public RpcImplementation getDefaultImplementation() {
  69         return defaultImplementation;
  70     }
  71
  72     public void setDefaultImplementation(RpcImplementation defaultImplementation) {
  73         this.defaultImplementation = defaultImplementation;
  74     }
  75
  76     public SchemaContextProvider getSchemaProvider() {
  77         return schemaProvider;
  78     }
  79
  80     public void setSchemaProvider(SchemaContextProvider schemaProvider) {
  81         this.schemaProvider = schemaProvider;
  82     }
  83
  84     @Override
  85     public RoutedRpcRegistration addRoutedRpcImplementation(QName rpcType, RpcImplementation implementation) {
  86         checkArgument(rpcType != null, "RPC Type should not be null");
  87         checkArgument(implementation != null, "RPC Implementatoin should not be null");
  88         return getOrCreateRoutedRpcRouter(rpcType).addRoutedRpcImplementation(rpcType, implementation);
  89     }
  90
  91     private RoutedRpcSelector getOrCreateRoutedRpcRouter(QName rpcType) {
  92         RoutedRpcSelector potential = getRoutedRpcRouter(rpcType);
  93         if (potential != null) {
  94             return potential;
  95         }
  96         synchronized (implementations) {
  97             potential = getRoutedRpcRouter(rpcType);
  98             if (potential != null) {
  99                 return potential;
 100             }
 101             RpcDefinition definition = findRpcDefinition(rpcType);
 102             RoutingStrategy strategy = getRoutingStrategy(definition);
 103             checkState(strategy instanceof RoutedRpcStrategy, "Rpc %s is not routed.", rpcType);
 104             potential = new RoutedRpcSelector((RoutedRpcStrategy) strategy, this);
 105             implementations.put(rpcType, potential);
 106             return potential;
 107         }
 108     }
 109
 110     private RoutedRpcSelector getRoutedRpcRouter(QName rpcType) {
 111         RpcImplementation potential = implementations.get(rpcType);
 112         if (potential != null) {
 113             checkState(potential instanceof RoutedRpcSelector, "Rpc %s is not routed.", rpcType);
 114             return (RoutedRpcSelector) potential;
 115         }
 116         return null;
 117
 118     }
 119
 120     @Override
 121     public RpcRegistration addRpcImplementation(QName rpcType, RpcImplementation implementation)
 122             throws IllegalArgumentException {
 123         checkArgument(rpcType != null, "RPC Type should not be null");
 124         checkArgument(implementation != null, "RPC Implementatoin should not be null");
 125         checkState(!hasRpcImplementation(rpcType), "Implementation already registered");
 126         RpcDefinition definition = findRpcDefinition(rpcType);
 127         checkArgument(!isRoutedRpc(definition), "RPC Type must not be routed.");
 128         GlobalRpcRegistration reg = new GlobalRpcRegistration(rpcType, implementation, this);
 129         return reg;
 130     }
 131
 132     private boolean isRoutedRpc(RpcDefinition definition) {
 133         return getRoutingStrategy(definition) instanceof RoutedRpcStrategy;
 134     }
 135
 136     @Override
 137     public ListenerRegistration<RpcRegistrationListener> addRpcRegistrationListener(RpcRegistrationListener listener) {
 138         return rpcRegistrationListeners.register(listener);
 139     }
 140
 141     @Override
 142     public String getIdentifier() {
 143         return identifier;
 144     }
 145
 146     @Override
 147     public Set<QName> getSupportedRpcs() {
 148         return ImmutableSet.copyOf(implementations.keySet());
 149     }
 150
 151     @Override
 152     public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
 153         return findRpcImplemention(rpc).invokeRpc(rpc, input);
 154     }
 155
 156     private RpcImplementation findRpcImplemention(QName rpc) {
 157         checkArgument(rpc != null, "Rpc name should not be null");
 158         RpcImplementation potentialImpl = implementations.get(rpc);
 159         if (potentialImpl != null) {
 160             return potentialImpl;
 161         }
 162         potentialImpl = defaultImplementation;
 163         checkState(potentialImpl != null, "Implementation is not available.");
 164         return potentialImpl;
 165     }
 166
 167     private boolean hasRpcImplementation(QName rpc) {
 168         return implementations.containsKey(rpc);
 169     }
 170
 171     private RpcDefinition findRpcDefinition(QName rpcType) {
 172         checkArgument(rpcType != null, "Rpc name must be supplied.");
 173         checkState(schemaProvider != null, "Schema Provider is not available.");
 174         SchemaContext ctx = schemaProvider.getSchemaContext();
 175         checkState(ctx != null, "YANG Schema Context is not available.");
 176         Module module = ctx.findModuleByNamespaceAndRevision(rpcType.getNamespace(), rpcType.getRevision());
 177         checkState(module != null, "YANG Module is not available.");
 178         return findRpcDefinition(rpcType, module.getRpcs());
 179     }
 180
 181     static private RpcDefinition findRpcDefinition(QName rpcType, Set<RpcDefinition> rpcs) {
 182         checkState(rpcs != null, "Rpc schema is not available.");
 183         for (RpcDefinition rpc : rpcs) {
 184             if (rpcType.equals(rpc.getQName())) {
 185                 return rpc;
 186             }
 187         }
 188         throw new IllegalArgumentException("Supplied Rpc Type is not defined.");
 189     }
 190
 191     private RoutingStrategy getRoutingStrategy(RpcDefinition rpc) {
 192         ContainerSchemaNode input = rpc.getInput();
 193         if (input != null) {
 194             for (DataSchemaNode schemaNode : input.getChildNodes()) {
 195                 Optional<QName> context = getRoutingContext(schemaNode);
 196                 if (context.isPresent()) {
 197                     return createRoutedStrategy(rpc, context.get(), schemaNode.getQName());
 198                 }
 199             }
 200         }
 201         return createGlobalStrategy(rpc);
 202     }
 203
 204     private static RoutingStrategy createRoutedStrategy(RpcDefinition rpc, QName context, QName leafNode) {
 205         return new RoutedRpcStrategy(rpc.getQName(), context, leafNode);
 206     }
 207
 208     private Optional<QName> getRoutingContext(DataSchemaNode schemaNode) {
 209         for (UnknownSchemaNode extension : schemaNode.getUnknownSchemaNodes()) {
 210             if (CONTEXT_REFERENCE.equals(extension.getNodeType())) {
 211                 return Optional.fromNullable(extension.getQName());
 212             }
 213             ;
 214         }
 215         return Optional.absent();
 216     }
 217
 218     private static RoutingStrategy createGlobalStrategy(RpcDefinition rpc) {
 219         GlobalRpcStrategy ret = new GlobalRpcStrategy(rpc.getQName());
 220         return ret;
 221     }
 222
 223     private static abstract class RoutingStrategy implements Identifiable<QName> {
 224
 225         private final QName identifier;
 226
 227         public RoutingStrategy(QName identifier) {
 228             super();
 229             this.identifier = identifier;
 230         }
 231
 232         @Override
 233         public QName getIdentifier() {
 234             return identifier;
 235         }
 236     }
 237
 238     private static class GlobalRpcStrategy extends RoutingStrategy {
 239
 240         public GlobalRpcStrategy(QName identifier) {
 241             super(identifier);
 242         }
 243     }
 244
 245     private static class RoutedRpcStrategy extends RoutingStrategy {
 246
 247         private final QName context;
 248         private final QName leaf;
 249
 250         public RoutedRpcStrategy(QName identifier, QName ctx, QName leaf) {
 251             super(identifier);
 252             this.context = ctx;
 253             this.leaf = leaf;
 254         }
 255
 256         public QName getContext() {
 257             return context;
 258         }
 259
 260         public QName getLeaf() {
 261             return leaf;
 262         }
 263     }
 264
 265     private static class RoutedRpcSelector implements RpcImplementation, AutoCloseable, Identifiable<QName> {
 266
 267         private final RoutedRpcStrategy strategy;
 268         private final Set<QName> supportedRpcs;
 269         private RpcImplementation defaultDelegate;
 270         private final ConcurrentMap<InstanceIdentifier, RoutedRpcRegImpl> implementations = new ConcurrentHashMap<>();
 271         private SchemaAwareRpcBroker router;
 272
 273         public RoutedRpcSelector(RoutedRpcStrategy strategy, SchemaAwareRpcBroker router) {
 274             super();
 275             this.strategy = strategy;
 276             supportedRpcs = ImmutableSet.of(strategy.getIdentifier());
 277             this.router = router;
 278         }
 279
 280         @Override
 281         public QName getIdentifier() {
 282             return strategy.getIdentifier();
 283         }
 284
 285         @Override
 286         public void close() throws Exception {
 287
 288         }
 289
 290         @Override
 291         public Set<QName> getSupportedRpcs() {
 292             return supportedRpcs;
 293         }
 294
 295         public RoutedRpcRegistration addRoutedRpcImplementation(QName rpcType, RpcImplementation implementation) {
 296             return new RoutedRpcRegImpl(rpcType, implementation, this);
 297         }
 298
 299         @Override
 300         public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
 301             CompositeNode inputContainer = input.getFirstCompositeByName(QName.create(rpc,"input"));
 302             checkArgument(inputContainer != null, "Rpc payload must contain input element");
 303             SimpleNode<?> routeContainer = inputContainer.getFirstSimpleByName(strategy.getLeaf());
 304             checkArgument(routeContainer != null, "Leaf %s must be set with value", strategy.getLeaf());
 305             Object route = routeContainer.getValue();
 306             RpcImplementation potential = null;
 307             if (route != null) {
 308                 RoutedRpcRegImpl potentialReg = implementations.get(route);
 309                 if (potentialReg != null) {
 310                     potential = potentialReg.getInstance();
 311                 }
 312             }
 313             if (potential == null) {
 314                 potential = defaultDelegate;
 315             }
 316             checkState(potential != null, "No implementation is available for rpc:%s path:%s", rpc, route);
 317             return potential.invokeRpc(rpc, input);
 318         }
 319
 320         public void addPath(QName context, InstanceIdentifier path, RoutedRpcRegImpl routedRpcRegImpl) {
 321             //checkArgument(strategy.getContext().equals(context),"Supplied context is not supported.");
 322             RoutedRpcRegImpl previous = implementations.put(path, routedRpcRegImpl);
 323             if (previous == null) {
 324                 router.notifyPathAnnouncement(context,strategy.getIdentifier(), path);
 325             }
 326
 327         }
 328
 329         public void removePath(QName context, InstanceIdentifier path, RoutedRpcRegImpl routedRpcRegImpl) {
 330             boolean removed = implementations.remove(path, routedRpcRegImpl);
 331             if (removed) {
 332                 router.notifyPathWithdrawal(context, strategy.getIdentifier(), path);
 333             }
 334         }
 335     }
 336
 337     private static class GlobalRpcRegistration extends AbstractObjectRegistration<RpcImplementation> implements
 338             RpcRegistration {
 339         private final QName type;
 340         private SchemaAwareRpcBroker router;
 341
 342         public GlobalRpcRegistration(QName type, RpcImplementation instance, SchemaAwareRpcBroker router) {
 343             super(instance);
 344             this.type = type;
 345             this.router = router;
 346         }
 347
 348         @Override
 349         public QName getType() {
 350             return type;
 351         }
 352
 353         @Override
 354         protected void removeRegistration() {
 355             if (router != null) {
 356                 router.remove(this);
 357                 router = null;
 358             }
 359         }
 360     }
 361
 362     private static class RoutedRpcRegImpl extends AbstractObjectRegistration<RpcImplementation> implements
 363             RoutedRpcRegistration {
 364
 365         private final QName type;
 366         private RoutedRpcSelector router;
 367
 368         public RoutedRpcRegImpl(QName rpcType, RpcImplementation implementation, RoutedRpcSelector routedRpcSelector) {
 369             super(implementation);
 370             this.type = rpcType;
 371             router = routedRpcSelector;
 372         }
 373
 374         @Override
 375         public void registerPath(QName context, InstanceIdentifier path) {
 376             router.addPath(context, path, this);
 377         }
 378
 379         @Override
 380         public void unregisterPath(QName context, InstanceIdentifier path) {
 381             router.removePath(context, path, this);
 382         }
 383
 384         @Override
 385         protected void removeRegistration() {
 386
 387         }
 388
 389         @Override
 390         public QName getType() {
 391             return type;
 392         }
 393
 394     }
 395
 396     private void remove(GlobalRpcRegistration registration) {
 397         implementations.remove(registration.getType(), registration);
 398     }
 399
 400     private void notifyPathAnnouncement(QName context, QName identifier, InstanceIdentifier path) {
 401         RpcRoutingContext contextWrapped = RpcRoutingContext.create(context, identifier);
 402         RouteChange<RpcRoutingContext, InstanceIdentifier> change = RoutingUtils.announcementChange(contextWrapped , path);
 403         for(ListenerRegistration<RouteChangeListener<RpcRoutingContext, InstanceIdentifier>> routeListener : routeChangeListeners) {
 404             try {
 405                 routeListener.getInstance().onRouteChange(change);
 406             } catch (Exception e) {
 407                 LOG.error("Unhandled exception during invoking onRouteChange for {}",routeListener.getInstance(),e);
 408
 409             }
 410         }
 411
 412     }
 413
 414
 415
 416     private void notifyPathWithdrawal(QName context,QName identifier, InstanceIdentifier path) {
 417         RpcRoutingContext contextWrapped = RpcRoutingContext.create(context, identifier);
 418         RouteChange<RpcRoutingContext, InstanceIdentifier> change = RoutingUtils.removalChange(contextWrapped , path);
 419         for(ListenerRegistration<RouteChangeListener<RpcRoutingContext, InstanceIdentifier>> routeListener : routeChangeListeners) {
 420             try {
 421                 routeListener.getInstance().onRouteChange(change);
 422             } catch (Exception e) {
 423                 LOG.error("Unhandled exception during invoking onRouteChange for {}",routeListener.getInstance(),e);
 424             }
 425         }
 426     }
 427
 428     @Override
 429     public <L extends RouteChangeListener<RpcRoutingContext, InstanceIdentifier>> ListenerRegistration<L> registerRouteChangeListener(
 430             L listener) {
 431         return routeChangeListeners.registerWithType(listener);
 432     }
 433 }