2 * Copyright (c) 2015 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
8 package org.opendaylight.controller.md.sal.dom.broker.impl;
10 import com.google.common.base.Function;
11 import com.google.common.base.Predicate;
12 import com.google.common.collect.Collections2;
13 import com.google.common.collect.ImmutableList;
14 import com.google.common.collect.ImmutableList.Builder;
15 import com.google.common.collect.ImmutableSet;
16 import com.google.common.util.concurrent.CheckedFuture;
17 import com.google.common.util.concurrent.ThreadFactoryBuilder;
18 import java.util.Collection;
19 import java.util.Collections;
21 import java.util.Map.Entry;
23 import java.util.concurrent.ExecutorService;
24 import java.util.concurrent.Executors;
25 import java.util.concurrent.ThreadFactory;
26 import javax.annotation.concurrent.GuardedBy;
27 import org.opendaylight.controller.md.sal.dom.api.DOMRpcAvailabilityListener;
28 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
29 import org.opendaylight.controller.md.sal.dom.api.DOMRpcIdentifier;
30 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementation;
31 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationRegistration;
32 import org.opendaylight.controller.md.sal.dom.api.DOMRpcProviderService;
33 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
34 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
35 import org.opendaylight.controller.md.sal.dom.spi.AbstractDOMRpcImplementationRegistration;
36 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
37 import org.opendaylight.yangtools.concepts.ListenerRegistration;
38 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
40 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
41 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
42 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
44 public final class DOMRpcRouter implements AutoCloseable, DOMRpcService, DOMRpcProviderService, SchemaContextListener {
45 private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder().setNameFormat("DOMRpcRouter-listener-%s").setDaemon(true).build();
46 private final ExecutorService listenerNotifier = Executors.newSingleThreadExecutor(THREAD_FACTORY);
48 private Collection<ListenerRegistration<? extends DOMRpcAvailabilityListener>> listeners = Collections.emptyList();
49 private volatile DOMRpcRoutingTable routingTable = DOMRpcRoutingTable.EMPTY;
52 public <T extends DOMRpcImplementation> DOMRpcImplementationRegistration<T> registerRpcImplementation(final T implementation, final DOMRpcIdentifier... rpcs) {
53 return registerRpcImplementation(implementation, ImmutableSet.copyOf(rpcs));
56 private static Collection<DOMRpcIdentifier> notPresentRpcs(final DOMRpcRoutingTable table, final Collection<DOMRpcIdentifier> candidates) {
57 return ImmutableSet.copyOf(Collections2.filter(candidates, new Predicate<DOMRpcIdentifier>() {
59 public boolean apply(final DOMRpcIdentifier input) {
60 return !table.contains(input);
65 private synchronized void removeRpcImplementation(final DOMRpcImplementation implementation, final Set<DOMRpcIdentifier> rpcs) {
66 final DOMRpcRoutingTable oldTable = routingTable;
67 final DOMRpcRoutingTable newTable = oldTable.remove(implementation, rpcs);
69 final Collection<DOMRpcIdentifier> removedRpcs = notPresentRpcs(newTable, rpcs);
70 final Collection<ListenerRegistration<? extends DOMRpcAvailabilityListener>> capturedListeners = listeners;
71 routingTable = newTable;
73 listenerNotifier.execute(new Runnable() {
76 for (ListenerRegistration<? extends DOMRpcAvailabilityListener> l : capturedListeners) {
77 // Need to ensure removed listeners do not get notified
78 synchronized (DOMRpcRouter.this) {
79 if (listeners.contains(l)) {
80 l.getInstance().onRpcUnavailable(removedRpcs);
89 public synchronized <T extends DOMRpcImplementation> DOMRpcImplementationRegistration<T> registerRpcImplementation(final T implementation, final Set<DOMRpcIdentifier> rpcs) {
90 final DOMRpcRoutingTable oldTable = routingTable;
91 final DOMRpcRoutingTable newTable = oldTable.add(implementation, rpcs);
93 final Collection<DOMRpcIdentifier> addedRpcs = notPresentRpcs(oldTable, rpcs);
94 final Collection<ListenerRegistration<? extends DOMRpcAvailabilityListener>> capturedListeners = listeners;
95 routingTable = newTable;
97 listenerNotifier.execute(new Runnable() {
100 for (ListenerRegistration<? extends DOMRpcAvailabilityListener> l : capturedListeners) {
101 // Need to ensure removed listeners do not get notified
102 synchronized (DOMRpcRouter.this) {
103 if (listeners.contains(l)) {
104 l.getInstance().onRpcAvailable(addedRpcs);
111 return new AbstractDOMRpcImplementationRegistration<T>(implementation) {
113 protected void removeRegistration() {
114 removeRpcImplementation(getInstance(), rpcs);
120 public CheckedFuture<DOMRpcResult, DOMRpcException> invokeRpc(final SchemaPath type, final NormalizedNode<?, ?> input) {
121 return routingTable.invokeRpc(type, input);
124 private synchronized void removeListener(final ListenerRegistration<? extends DOMRpcAvailabilityListener> reg) {
125 listeners = ImmutableList.copyOf(Collections2.filter(listeners, new Predicate<Object>() {
127 public boolean apply(final Object input) {
128 return !reg.equals(input);
134 public synchronized <T extends DOMRpcAvailabilityListener> ListenerRegistration<T> registerRpcListener(final T listener) {
135 final ListenerRegistration<T> ret = new AbstractListenerRegistration<T>(listener) {
137 protected void removeRegistration() {
138 removeListener(this);
142 final Builder<ListenerRegistration<? extends DOMRpcAvailabilityListener>> b = ImmutableList.builder();
145 listeners = b.build();
146 final Map<SchemaPath, Set<YangInstanceIdentifier>> capturedRpcs = routingTable.getRpcs();
148 listenerNotifier.execute(new Runnable() {
151 for (final Entry<SchemaPath, Set<YangInstanceIdentifier>> e : capturedRpcs.entrySet()) {
152 listener.onRpcAvailable(Collections2.transform(e.getValue(), new Function<YangInstanceIdentifier, DOMRpcIdentifier>() {
154 public DOMRpcIdentifier apply(final YangInstanceIdentifier input) {
155 return DOMRpcIdentifier.create(e.getKey(), input);
166 public synchronized void onGlobalContextUpdated(final SchemaContext context) {
167 final DOMRpcRoutingTable oldTable = routingTable;
168 final DOMRpcRoutingTable newTable = oldTable.setSchemaContext(context);
169 routingTable = newTable;
173 public void close() {
174 listenerNotifier.shutdown();