dom/mdsal-dom-broker/src/main/java/org/opendaylight/mdsal/dom/broker/DOMRpcRouter.java

   1 /*
   2  * Copyright (c) 2015 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.mdsal.dom.broker;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Verify.verifyNotNull;
  12 import static java.util.Objects.requireNonNull;
  13
  14 import com.google.common.annotations.VisibleForTesting;
  15 import com.google.common.collect.Collections2;
  16 import com.google.common.collect.ImmutableList;
  17 import com.google.common.collect.ImmutableSet;
  18 import com.google.common.collect.ImmutableTable;
  19 import com.google.common.collect.MapDifference;
  20 import com.google.common.collect.MapDifference.ValueDifference;
  21 import com.google.common.collect.Maps;
  22 import com.google.common.collect.Sets;
  23 import com.google.common.util.concurrent.Futures;
  24 import com.google.common.util.concurrent.ListenableFuture;
  25 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  26 import java.util.ArrayList;
  27 import java.util.Collection;
  28 import java.util.HashSet;
  29 import java.util.List;
  30 import java.util.Map;
  31 import java.util.Map.Entry;
  32 import java.util.Set;
  33 import java.util.concurrent.ExecutorService;
  34 import java.util.concurrent.Executors;
  35 import java.util.concurrent.ThreadFactory;
  36 import javax.annotation.PreDestroy;
  37 import javax.inject.Inject;
  38 import javax.inject.Singleton;
  39 import org.checkerframework.checker.lock.qual.GuardedBy;
  40 import org.eclipse.jdt.annotation.NonNull;
  41 import org.eclipse.jdt.annotation.NonNullByDefault;
  42 import org.opendaylight.mdsal.dom.api.DOMActionAvailabilityExtension;
  43 import org.opendaylight.mdsal.dom.api.DOMActionAvailabilityExtension.AvailabilityListener;
  44 import org.opendaylight.mdsal.dom.api.DOMActionImplementation;
  45 import org.opendaylight.mdsal.dom.api.DOMActionInstance;
  46 import org.opendaylight.mdsal.dom.api.DOMActionNotAvailableException;
  47 import org.opendaylight.mdsal.dom.api.DOMActionProviderService;
  48 import org.opendaylight.mdsal.dom.api.DOMActionResult;
  49 import org.opendaylight.mdsal.dom.api.DOMActionService;
  50 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  51 import org.opendaylight.mdsal.dom.api.DOMRpcAvailabilityListener;
  52 import org.opendaylight.mdsal.dom.api.DOMRpcIdentifier;
  53 import org.opendaylight.mdsal.dom.api.DOMRpcImplementation;
  54 import org.opendaylight.mdsal.dom.api.DOMRpcImplementationNotAvailableException;
  55 import org.opendaylight.mdsal.dom.api.DOMRpcProviderService;
  56 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  57 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  58 import org.opendaylight.mdsal.dom.api.DOMSchemaService;
  59 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
  60 import org.opendaylight.yangtools.concepts.AbstractRegistration;
  61 import org.opendaylight.yangtools.concepts.ObjectRegistration;
  62 import org.opendaylight.yangtools.concepts.Registration;
  63 import org.opendaylight.yangtools.yang.common.QName;
  64 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  65 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  66 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  67 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContextListener;
  68 import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
  69 import org.osgi.service.component.annotations.Activate;
  70 import org.osgi.service.component.annotations.Component;
  71 import org.osgi.service.component.annotations.Deactivate;
  72 import org.osgi.service.component.annotations.Reference;
  73 import org.slf4j.Logger;
  74 import org.slf4j.LoggerFactory;
  75
  76 @Singleton
  77 @Component(service = DOMRpcRouter.class)
  78 public final class DOMRpcRouter extends AbstractRegistration implements EffectiveModelContextListener {
  79     private static final Logger LOG = LoggerFactory.getLogger(DOMRpcRouter.class);
  80     private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder().setNameFormat(
  81             "DOMRpcRouter-listener-%s").setDaemon(true).build();
  82
  83     private final ExecutorService listenerNotifier = Executors.newSingleThreadExecutor(THREAD_FACTORY);
  84     private final @NonNull DOMActionProviderService actionProviderService = new ActionProviderServiceFacade();
  85     private final @NonNull DOMActionService actionService = new ActionServiceFacade();
  86     private final @NonNull DOMRpcProviderService rpcProviderService = new RpcProviderServiceFacade();
  87     private final @NonNull DOMRpcService rpcService = new RpcServiceFacade();
  88
  89     @GuardedBy("this")
  90     private ImmutableList<RegImpl> listeners = ImmutableList.of();
  91
  92     @GuardedBy("this")
  93     private ImmutableList<ActionRegistration> actionListeners = ImmutableList.of();
  94
  95     private volatile DOMRpcRoutingTable routingTable = DOMRpcRoutingTable.EMPTY;
  96
  97     private volatile DOMActionRoutingTable actionRoutingTable = DOMActionRoutingTable.EMPTY;
  98
  99     private Registration listenerRegistration;
 100
 101     @Deprecated
 102     @VisibleForTesting
 103     // FIXME: 9.0.0: make this constructor package-private
 104     public DOMRpcRouter() {
 105
 106     }
 107
 108     @Inject
 109     @Activate
 110     public DOMRpcRouter(@Reference final DOMSchemaService schemaService) {
 111         listenerRegistration = schemaService.registerSchemaContextListener(this);
 112         LOG.info("DOM RPC/Action router started");
 113     }
 114
 115     @Deprecated(forRemoval = true)
 116     public static DOMRpcRouter newInstance(final DOMSchemaService schemaService) {
 117         return new DOMRpcRouter(schemaService);
 118     }
 119
 120     @PreDestroy
 121     @Deactivate
 122     public void shutdown() {
 123         close();
 124     }
 125
 126     public @NonNull DOMActionService actionService() {
 127         return actionService;
 128     }
 129
 130     public @NonNull DOMActionProviderService actionProviderService() {
 131         return actionProviderService;
 132     }
 133
 134     public @NonNull DOMRpcService rpcService() {
 135         return rpcService;
 136     }
 137
 138     public @NonNull DOMRpcProviderService rpcProviderService() {
 139         return rpcProviderService;
 140     }
 141
 142     private synchronized void removeRpcImplementation(final DOMRpcImplementation implementation,
 143             final Set<DOMRpcIdentifier> rpcs) {
 144         final DOMRpcRoutingTable oldTable = routingTable;
 145         final DOMRpcRoutingTable newTable = (DOMRpcRoutingTable) oldTable.remove(implementation, rpcs);
 146         routingTable = newTable;
 147
 148         listenerNotifier.execute(() -> notifyRemoved(newTable, implementation));
 149     }
 150
 151     private synchronized void removeRpcImplementations(
 152             final ImmutableTable<QName, YangInstanceIdentifier, DOMRpcImplementation> implTable) {
 153         final DOMRpcRoutingTable oldTable = routingTable;
 154         final DOMRpcRoutingTable newTable = (DOMRpcRoutingTable) oldTable.removeAll(implTable);
 155         routingTable = newTable;
 156
 157         listenerNotifier.execute(() -> notifyRemoved(newTable, implTable.values()));
 158     }
 159
 160     private synchronized void removeActionImplementation(final DOMActionImplementation implementation,
 161             final Set<DOMActionInstance> actions) {
 162         final DOMActionRoutingTable oldTable = actionRoutingTable;
 163         final DOMActionRoutingTable newTable = (DOMActionRoutingTable) oldTable.remove(implementation, actions);
 164         actionRoutingTable = newTable;
 165
 166         listenerNotifier.execute(() -> notifyActionChanged(newTable, implementation));
 167     }
 168
 169     private synchronized void removeListener(final RegImpl reg) {
 170         listeners = ImmutableList.copyOf(Collections2.filter(listeners, input -> !reg.equals(input)));
 171     }
 172
 173     private synchronized void removeActionListener(final ActionRegistration reg) {
 174         actionListeners = ImmutableList.copyOf(Collections2.filter(actionListeners, input -> !reg.equals(input)));
 175     }
 176
 177     private synchronized void notifyAdded(final DOMRpcRoutingTable newTable, final DOMRpcImplementation impl) {
 178         for (var l : listeners) {
 179             l.addRpc(newTable, impl);
 180         }
 181     }
 182
 183     private synchronized void notifyAdded(final DOMRpcRoutingTable newTable,
 184             final Collection<? extends DOMRpcImplementation> impls) {
 185         for (var l : listeners) {
 186             for (var impl : impls) {
 187                 l.addRpc(newTable, impl);
 188             }
 189         }
 190     }
 191
 192     private synchronized void notifyRemoved(final DOMRpcRoutingTable newTable, final DOMRpcImplementation impl) {
 193         for (var l : listeners) {
 194             l.removeRpc(newTable, impl);
 195         }
 196     }
 197
 198     private synchronized void notifyRemoved(final DOMRpcRoutingTable newTable,
 199             final Collection<? extends DOMRpcImplementation> impls) {
 200         for (var l : listeners) {
 201             for (DOMRpcImplementation impl : impls) {
 202                 l.removeRpc(newTable, impl);
 203             }
 204         }
 205     }
 206
 207     private synchronized void notifyActionChanged(final DOMActionRoutingTable newTable,
 208             final DOMActionImplementation impl) {
 209         for (var l : actionListeners) {
 210             l.actionChanged(newTable, impl);
 211         }
 212     }
 213
 214     @Override
 215     public synchronized void onModelContextUpdated(final EffectiveModelContext newModelContext) {
 216         final DOMRpcRoutingTable oldTable = routingTable;
 217         final DOMRpcRoutingTable newTable = (DOMRpcRoutingTable) oldTable.setSchemaContext(newModelContext);
 218         routingTable = newTable;
 219
 220         final DOMActionRoutingTable oldActionTable = actionRoutingTable;
 221         final DOMActionRoutingTable newActionTable =
 222                 (DOMActionRoutingTable) oldActionTable.setSchemaContext(newModelContext);
 223         actionRoutingTable = newActionTable;
 224     }
 225
 226     @Override
 227     protected void removeRegistration() {
 228         if (listenerRegistration != null) {
 229             listenerRegistration.close();
 230             listenerRegistration = null;
 231         }
 232         listenerNotifier.shutdown();
 233         LOG.info("DOM RPC/Action router stopped");
 234     }
 235
 236     @VisibleForTesting
 237     synchronized List<?> listeners() {
 238         return listeners;
 239     }
 240
 241     @VisibleForTesting
 242     synchronized List<?> actionListeners() {
 243         return actionListeners;
 244     }
 245
 246     @VisibleForTesting
 247     DOMRpcRoutingTable routingTable() {
 248         return routingTable;
 249     }
 250
 251     private static final class RegImpl extends AbstractObjectRegistration<DOMRpcAvailabilityListener> {
 252         private Map<QName, Set<YangInstanceIdentifier>> prevRpcs;
 253         private DOMRpcRouter router;
 254
 255         RegImpl(final DOMRpcRouter router, final DOMRpcAvailabilityListener listener,
 256                 final Map<QName, Set<YangInstanceIdentifier>> rpcs) {
 257             super(listener);
 258             this.router = requireNonNull(router);
 259             prevRpcs = requireNonNull(rpcs);
 260         }
 261
 262         @Override
 263         protected void removeRegistration() {
 264             router.removeListener(this);
 265             router = null;
 266         }
 267
 268         void initialTable() {
 269             final List<DOMRpcIdentifier> added = new ArrayList<>();
 270             for (Entry<QName, Set<YangInstanceIdentifier>> e : prevRpcs.entrySet()) {
 271                 added.addAll(Collections2.transform(e.getValue(), i -> DOMRpcIdentifier.create(e.getKey(), i)));
 272             }
 273             if (!added.isEmpty()) {
 274                 getInstance().onRpcAvailable(added);
 275             }
 276         }
 277
 278         void addRpc(final DOMRpcRoutingTable newTable, final DOMRpcImplementation impl) {
 279             final var l = getInstance();
 280             if (!l.acceptsImplementation(impl)) {
 281                 return;
 282             }
 283
 284             final Map<QName, Set<YangInstanceIdentifier>> rpcs = verifyNotNull(newTable.getOperations(l));
 285             final MapDifference<QName, Set<YangInstanceIdentifier>> diff = Maps.difference(prevRpcs, rpcs);
 286
 287             final List<DOMRpcIdentifier> added = new ArrayList<>();
 288             for (Entry<QName, Set<YangInstanceIdentifier>> e : diff.entriesOnlyOnRight().entrySet()) {
 289                 added.addAll(Collections2.transform(e.getValue(), i -> DOMRpcIdentifier.create(e.getKey(), i)));
 290             }
 291             for (Entry<QName, ValueDifference<Set<YangInstanceIdentifier>>> e : diff.entriesDiffering().entrySet()) {
 292                 for (YangInstanceIdentifier i : Sets.difference(e.getValue().rightValue(), e.getValue().leftValue())) {
 293                     added.add(DOMRpcIdentifier.create(e.getKey(), i));
 294                 }
 295             }
 296
 297             prevRpcs = rpcs;
 298             if (!added.isEmpty()) {
 299                 l.onRpcAvailable(added);
 300             }
 301         }
 302
 303         void removeRpc(final DOMRpcRoutingTable newTable, final DOMRpcImplementation impl) {
 304             final var l = getInstance();
 305             if (!l.acceptsImplementation(impl)) {
 306                 return;
 307             }
 308
 309             final Map<QName, Set<YangInstanceIdentifier>> rpcs = verifyNotNull(newTable.getOperations(l));
 310             final MapDifference<QName, Set<YangInstanceIdentifier>> diff = Maps.difference(prevRpcs, rpcs);
 311
 312             final List<DOMRpcIdentifier> removed = new ArrayList<>();
 313             for (Entry<QName, Set<YangInstanceIdentifier>> e : diff.entriesOnlyOnLeft().entrySet()) {
 314                 removed.addAll(Collections2.transform(e.getValue(), i -> DOMRpcIdentifier.create(e.getKey(), i)));
 315             }
 316             for (Entry<QName, ValueDifference<Set<YangInstanceIdentifier>>> e : diff.entriesDiffering().entrySet()) {
 317                 for (YangInstanceIdentifier i : Sets.difference(e.getValue().leftValue(), e.getValue().rightValue())) {
 318                     removed.add(DOMRpcIdentifier.create(e.getKey(), i));
 319                 }
 320             }
 321
 322             prevRpcs = rpcs;
 323             if (!removed.isEmpty()) {
 324                 l.onRpcUnavailable(removed);
 325             }
 326         }
 327     }
 328
 329     private static final class ActionRegistration extends AbstractObjectRegistration<AvailabilityListener> {
 330         private Map<Absolute, Set<DOMDataTreeIdentifier>> prevActions;
 331         private DOMRpcRouter router;
 332
 333         ActionRegistration(final DOMRpcRouter router, final AvailabilityListener listener,
 334                 final Map<Absolute, Set<DOMDataTreeIdentifier>> actions) {
 335             super(listener);
 336             this.router = requireNonNull(router);
 337             prevActions = requireNonNull(actions);
 338         }
 339
 340         @Override
 341         protected void removeRegistration() {
 342             router.removeActionListener(this);
 343             router = null;
 344         }
 345
 346         void initialTable() {
 347             final var added = new ArrayList<DOMActionInstance>();
 348             for (var e : prevActions.entrySet()) {
 349                 added.addAll(Collections2.transform(e.getValue(), i -> DOMActionInstance.of(e.getKey(), i)));
 350             }
 351             if (!added.isEmpty()) {
 352                 getInstance().onActionsChanged(ImmutableSet.of(), ImmutableSet.copyOf(added));
 353             }
 354         }
 355
 356         void actionChanged(final DOMActionRoutingTable newTable, final DOMActionImplementation impl) {
 357             final var l = getInstance();
 358             if (!l.acceptsImplementation(impl)) {
 359                 return;
 360             }
 361
 362             final Map<Absolute, Set<DOMDataTreeIdentifier>> actions = verifyNotNull(newTable.getOperations(l));
 363             final MapDifference<Absolute, Set<DOMDataTreeIdentifier>> diff = Maps.difference(prevActions, actions);
 364
 365             final Set<DOMActionInstance> removed = new HashSet<>();
 366             final Set<DOMActionInstance> added = new HashSet<>();
 367
 368             for (Entry<Absolute, Set<DOMDataTreeIdentifier>> e : diff.entriesOnlyOnLeft().entrySet()) {
 369                 removed.addAll(Collections2.transform(e.getValue(), i -> DOMActionInstance.of(e.getKey(), i)));
 370             }
 371
 372             for (Entry<Absolute, Set<DOMDataTreeIdentifier>> e : diff.entriesOnlyOnRight().entrySet()) {
 373                 added.addAll(Collections2.transform(e.getValue(), i -> DOMActionInstance.of(e.getKey(), i)));
 374             }
 375
 376             for (Entry<Absolute, ValueDifference<Set<DOMDataTreeIdentifier>>> e : diff.entriesDiffering().entrySet()) {
 377                 for (DOMDataTreeIdentifier i : Sets.difference(e.getValue().leftValue(), e.getValue().rightValue())) {
 378                     removed.add(DOMActionInstance.of(e.getKey(), i));
 379                 }
 380
 381                 for (DOMDataTreeIdentifier i : Sets.difference(e.getValue().rightValue(), e.getValue().leftValue())) {
 382                     added.add(DOMActionInstance.of(e.getKey(), i));
 383                 }
 384             }
 385
 386             prevActions = actions;
 387             if (!removed.isEmpty() || !added.isEmpty()) {
 388                 l.onActionsChanged(removed, added);
 389             }
 390         }
 391     }
 392
 393     @NonNullByDefault
 394     private final class ActionServiceFacade implements DOMActionService, DOMActionAvailabilityExtension {
 395         @Override
 396         public List<Extension> supportedExtensions() {
 397             return List.of(this);
 398         }
 399
 400         @Override
 401         public ListenableFuture<? extends DOMActionResult> invokeAction(final Absolute type,
 402                 final DOMDataTreeIdentifier path, final ContainerNode input) {
 403             final YangInstanceIdentifier pathRoot = path.path();
 404             checkArgument(!pathRoot.isEmpty(), "Action path must not be empty");
 405
 406             final DOMActionRoutingTableEntry entry = (DOMActionRoutingTableEntry) actionRoutingTable.getEntry(type);
 407             return entry != null ? OperationInvocation.invoke(entry, type, path, requireNonNull(input))
 408                 : Futures.immediateFailedFuture(
 409                     new DOMActionNotAvailableException("No implementation of Action %s available", type));
 410         }
 411
 412         @Override
 413         public Registration registerAvailabilityListener(final AvailabilityListener listener) {
 414             synchronized (DOMRpcRouter.this) {
 415                 final var ret = new ActionRegistration(DOMRpcRouter.this, listener,
 416                     actionRoutingTable.getOperations(listener));
 417                 actionListeners = ImmutableList.<ActionRegistration>builder()
 418                     .addAll(actionListeners)
 419                     .add(ret)
 420                     .build();
 421
 422                 listenerNotifier.execute(ret::initialTable);
 423                 return ret;
 424             }
 425         }
 426     }
 427
 428     @NonNullByDefault
 429     private final class ActionProviderServiceFacade implements DOMActionProviderService {
 430         @Override
 431         public <T extends DOMActionImplementation> ObjectRegistration<T> registerActionImplementation(
 432                 final T implementation, final Set<DOMActionInstance> instances) {
 433             checkArgument(!instances.isEmpty(), "Instances must not be empty");
 434
 435             synchronized (DOMRpcRouter.this) {
 436                 final DOMActionRoutingTable oldTable = actionRoutingTable;
 437                 final DOMActionRoutingTable newTable = (DOMActionRoutingTable) oldTable.add(implementation, instances);
 438                 actionRoutingTable = newTable;
 439
 440                 listenerNotifier.execute(() -> notifyActionChanged(newTable, implementation));
 441             }
 442
 443             return new AbstractObjectRegistration<>(implementation) {
 444                 @Override
 445                 protected void removeRegistration() {
 446                     removeActionImplementation(getInstance(), instances);
 447                 }
 448             };
 449         }
 450     }
 451
 452     private final class RpcServiceFacade implements DOMRpcService {
 453         @Override
 454         public ListenableFuture<? extends DOMRpcResult> invokeRpc(final QName type, final ContainerNode input) {
 455             final var entry = (AbstractDOMRpcRoutingTableEntry) routingTable.getEntry(type);
 456             if (entry == null) {
 457                 return Futures.immediateFailedFuture(
 458                     new DOMRpcImplementationNotAvailableException("No implementation of RPC %s available", type));
 459             }
 460
 461             return OperationInvocation.invoke(entry, requireNonNull(input));
 462         }
 463
 464         @Override
 465         public Registration registerRpcListener(final DOMRpcAvailabilityListener listener) {
 466             synchronized (DOMRpcRouter.this) {
 467                 final var ret = new RegImpl(DOMRpcRouter.this, listener, routingTable.getOperations(listener));
 468                 listeners = ImmutableList.<RegImpl>builder().addAll(listeners).add(ret).build();
 469
 470                 listenerNotifier.execute(ret::initialTable);
 471                 return ret;
 472             }
 473         }
 474     }
 475
 476     private final class RpcProviderServiceFacade implements DOMRpcProviderService {
 477         @Override
 478         public Registration registerRpcImplementation(final DOMRpcImplementation implementation,
 479                 final Set<DOMRpcIdentifier> rpcs) {
 480
 481             synchronized (DOMRpcRouter.this) {
 482                 final DOMRpcRoutingTable oldTable = routingTable;
 483                 final DOMRpcRoutingTable newTable = (DOMRpcRoutingTable) oldTable.add(implementation, rpcs);
 484                 routingTable = newTable;
 485
 486                 listenerNotifier.execute(() -> notifyAdded(newTable, implementation));
 487             }
 488
 489             return new AbstractRegistration() {
 490                 @Override
 491                 protected void removeRegistration() {
 492                     removeRpcImplementation(implementation, rpcs);
 493                 }
 494             };
 495         }
 496
 497         @Override
 498         public Registration registerRpcImplementations(final Map<DOMRpcIdentifier, DOMRpcImplementation> map) {
 499             checkArgument(!map.isEmpty());
 500
 501             final var builder = ImmutableTable.<QName, YangInstanceIdentifier, DOMRpcImplementation>builder();
 502             for (var entry : map.entrySet()) {
 503                 final var id = entry.getKey();
 504                 builder.put(id.getType(), id.getContextReference(), entry.getValue());
 505             }
 506             final var implTable = builder.build();
 507
 508             synchronized (DOMRpcRouter.this) {
 509                 final DOMRpcRoutingTable oldTable = routingTable;
 510                 final DOMRpcRoutingTable newTable = (DOMRpcRoutingTable) oldTable.addAll(implTable);
 511                 routingTable = newTable;
 512
 513                 listenerNotifier.execute(() -> notifyAdded(newTable, implTable.values()));
 514             }
 515
 516             return new AbstractRegistration() {
 517                 @Override
 518                 protected void removeRegistration() {
 519                     removeRpcImplementations(implTable);
 520                 }
 521             };
 522         }
 523     }
 524 }