import org.opendaylight.controller.config.api.DependencyResolver;
import org.opendaylight.controller.config.api.ModuleIdentifier;
-import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodec;
import org.opendaylight.controller.md.sal.binding.impl.BindingDOMNotificationServiceAdapter;
+import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodec;
import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
-import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
import org.opendaylight.controller.sal.core.api.Broker;
public class BindingNotificationAdapterModule extends AbstractBindingNotificationAdapterModule {
- public BindingNotificationAdapterModule(ModuleIdentifier identifier, DependencyResolver dependencyResolver) {
+ public BindingNotificationAdapterModule(final ModuleIdentifier identifier, final DependencyResolver dependencyResolver) {
super(identifier, dependencyResolver);
}
- public BindingNotificationAdapterModule(ModuleIdentifier identifier, DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.md.sal.binding.impl.BindingNotificationAdapterModule oldModule, java.lang.AutoCloseable oldInstance) {
+ public BindingNotificationAdapterModule(final ModuleIdentifier identifier, final DependencyResolver dependencyResolver, final org.opendaylight.controller.config.yang.md.sal.binding.impl.BindingNotificationAdapterModule oldModule, final java.lang.AutoCloseable oldInstance) {
super(identifier, dependencyResolver, oldModule, oldInstance);
}
final BindingToNormalizedNodeCodec codec = getBindingMappingServiceDependency();
final Broker.ProviderSession session = getDomAsyncBrokerDependency().registerProvider(new DummyDOMProvider());
final DOMNotificationService notifService = session.getService(DOMNotificationService.class);
- return new BindingDOMNotificationServiceAdapter(codec.getCodecRegistry(), notifService, SingletonHolder.INVOKER_FACTORY);
+ return new BindingDOMNotificationServiceAdapter(codec.getCodecRegistry(), notifService);
}
}
*/
package org.opendaylight.controller.config.yang.md.sal.binding.impl;
-import com.google.common.util.concurrent.ListeningExecutorService;
+import org.opendaylight.controller.md.sal.binding.compat.HydrogenNotificationBrokerImpl;
+
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
+import org.opendaylight.controller.md.sal.binding.api.NotificationService;
+import org.opendaylight.controller.md.sal.binding.compat.HeliumNotificationProviderServiceAdapter;
import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
-import org.opendaylight.controller.sal.binding.impl.NotificationBrokerImpl;
/**
*
public final class NotificationBrokerImplModule extends
org.opendaylight.controller.config.yang.md.sal.binding.impl.AbstractNotificationBrokerImplModule {
- public NotificationBrokerImplModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier,
- org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ public NotificationBrokerImplModule(final org.opendaylight.controller.config.api.ModuleIdentifier identifier,
+ final org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
super(identifier, dependencyResolver);
}
- public NotificationBrokerImplModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier,
- org.opendaylight.controller.config.api.DependencyResolver dependencyResolver,
- NotificationBrokerImplModule oldModule, java.lang.AutoCloseable oldInstance) {
+ public NotificationBrokerImplModule(final org.opendaylight.controller.config.api.ModuleIdentifier identifier,
+ final org.opendaylight.controller.config.api.DependencyResolver dependencyResolver,
+ final NotificationBrokerImplModule oldModule, final java.lang.AutoCloseable oldInstance) {
super(identifier, dependencyResolver, oldModule, oldInstance);
}
@Override
public java.lang.AutoCloseable createInstance() {
+ final NotificationPublishService notificationPublishService = getNotificationPublishAdapterDependency();
+ final NotificationService notificationService = getNotificationAdapterDependency();
+
+ if(notificationPublishService != null & notificationService != null) {
+ return new HeliumNotificationProviderServiceAdapter(notificationPublishService, notificationService);
+ }
+
/*
* FIXME: Switch to new broker (which has different threading model)
* once this change is communicated with downstream users or
* we will have adapter implementation which will honor Helium
* threading model for notifications.
*/
- ListeningExecutorService listeningExecutor = SingletonHolder.getDefaultNotificationExecutor();
- NotificationBrokerImpl broker = new NotificationBrokerImpl(listeningExecutor);
- return broker;
+
+ return new HydrogenNotificationBrokerImpl(SingletonHolder.getDefaultNotificationExecutor());
}
}
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.controller.sal.binding.impl;
+package org.opendaylight.controller.md.sal.binding.compat;
import org.opendaylight.controller.sal.binding.api.NotificationListener;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.controller.sal.binding.impl;
+package org.opendaylight.controller.md.sal.binding.compat;
+import com.google.common.base.Preconditions;
import org.opendaylight.controller.sal.binding.api.NotificationListener;
import org.opendaylight.yangtools.yang.binding.Notification;
-import com.google.common.base.Preconditions;
-
/**
* An aggregated listener registration. This is a result of registering an invoker which can handle multiple
* interfaces at the same time. In order to support correct delivery, we need to maintain per-type registrations
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.controller.sal.binding.impl;
+package org.opendaylight.controller.md.sal.binding.compat;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicReference;
-
import javax.annotation.concurrent.GuardedBy;
-
import org.opendaylight.controller.sal.binding.api.NotificationListener;
import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
-import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.util.ListenerRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.Multimap;
-
-public class NotificationBrokerImpl implements NotificationProviderService, AutoCloseable {
- private static final Logger LOG = LoggerFactory.getLogger(NotificationBrokerImpl.class);
+@Deprecated
+public class HydrogenNotificationBrokerImpl implements NotificationProviderService, AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(HydrogenNotificationBrokerImpl.class);
private final ListenerRegistry<NotificationInterestListener> interestListeners =
ListenerRegistry.create();
private final AtomicReference<ListenerMapGeneration> listeners = new AtomicReference<>(new ListenerMapGeneration());
private final ExecutorService executor;
- public NotificationBrokerImpl(final ExecutorService executor) {
+ public HydrogenNotificationBrokerImpl(final ExecutorService executor) {
this.executor = Preconditions.checkNotNull(executor);
}
@Override
public void publish(final Notification notification, final ExecutorService service) {
- for (NotificationListenerRegistration<?> r : listeners.get().listenersFor(notification)) {
+ for (final NotificationListenerRegistration<?> r : listeners.get().listenersFor(notification)) {
service.submit(new NotifyTask(r, notification));
}
}
synchronized (this) {
final Multimap<Class<? extends Notification>, NotificationListenerRegistration<?>> newListeners =
mutableListeners();
- for (NotificationListenerRegistration<?> reg : registrations) {
+ for (final NotificationListenerRegistration<?> reg : registrations) {
newListeners.put(reg.getType(), reg);
}
}
// Notifications are dispatched out of lock...
- for (NotificationListenerRegistration<?> reg : registrations) {
+ for (final NotificationListenerRegistration<?> reg : registrations) {
announceNotificationSubscription(reg.getType());
}
}
final Multimap<Class<? extends Notification>, NotificationListenerRegistration<?>> newListeners =
mutableListeners();
- for (NotificationListenerRegistration<?> reg : registrations) {
+ for (final NotificationListenerRegistration<?> reg : registrations) {
newListeners.remove(reg.getType(), reg);
}
for (final ListenerRegistration<NotificationInterestListener> listener : interestListeners) {
try {
listener.getInstance().onNotificationSubscribtion(notification);
- } catch (Exception e) {
+ } catch (final Exception e) {
LOG.warn("Listener {} reported unexpected error on notification {}",
listener.getInstance(), notification, e);
}
@Override
public ListenerRegistration<org.opendaylight.yangtools.yang.binding.NotificationListener> registerNotificationListener(final org.opendaylight.yangtools.yang.binding.NotificationListener listener) {
- final NotificationInvoker invoker = SingletonHolder.INVOKER_FACTORY.invokerFor(listener);
+ final NotificationInvoker invoker = NotificationInvoker.invokerFor(listener);
final Set<Class<? extends Notification>> types = invoker.getSupportedNotifications();
final NotificationListenerRegistration<?>[] regs = new NotificationListenerRegistration<?>[types.size()];
// Populate the registrations...
int i = 0;
- for (Class<? extends Notification> type : types) {
- regs[i] = new AggregatedNotificationListenerRegistration<Notification, Object>(type, invoker.getInvocationProxy(), regs) {
+ for (final Class<? extends Notification> type : types) {
+ regs[i] = new AggregatedNotificationListenerRegistration<Notification, Object>(type, invoker, regs) {
@Override
protected void removeRegistration() {
// Nothing to do, will be cleaned up by parent (below)
@Override
protected void removeRegistration() {
removeRegistrations(regs);
- for (ListenerRegistration<?> reg : regs) {
+ for (final ListenerRegistration<?> reg : regs) {
reg.close();
}
}
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.controller.sal.binding.impl;
+package org.opendaylight.controller.md.sal.binding.compat;
import java.util.Arrays;
import java.util.Collection;
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.md.sal.binding.compat;
+
+import com.google.common.collect.ImmutableMap;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.opendaylight.yangtools.yang.binding.Notification;
+import org.opendaylight.yangtools.yang.binding.NotificationListener;
+import org.opendaylight.yangtools.yang.binding.util.BindingReflections;
+import org.opendaylight.yangtools.yang.binding.util.NotificationListenerInvoker;
+import org.opendaylight.yangtools.yang.common.QName;
+
+final class NotificationInvoker implements org.opendaylight.controller.sal.binding.api.NotificationListener<Notification> {
+
+ private final NotificationListener delegate;
+ private final Map<Class<? extends Notification>,InvokerContext> invokers;
+
+
+ private NotificationInvoker(final NotificationListener listener) {
+ delegate = listener;
+ final Map<Class<? extends Notification>, InvokerContext> builder = new HashMap<>();
+ for(final Class<?> iface : listener.getClass().getInterfaces()) {
+ if(NotificationListener.class.isAssignableFrom(iface) && BindingReflections.isBindingClass(iface)) {
+ @SuppressWarnings("unchecked")
+ final Class<? extends NotificationListener> listenerType = (Class<? extends NotificationListener>) iface;
+ final NotificationListenerInvoker invoker = NotificationListenerInvoker.from(listenerType);
+ for(final Class<? extends Notification> type : getNotificationTypes(listenerType)) {
+ builder.put(type, new InvokerContext(BindingReflections.findQName(type) , invoker));
+ }
+ }
+ }
+ invokers = ImmutableMap.copyOf(builder);
+ }
+
+ public static NotificationInvoker invokerFor(final NotificationListener listener) {
+ return new NotificationInvoker(listener);
+ }
+
+ public Set<Class<? extends Notification>> getSupportedNotifications() {
+ return invokers.keySet();
+ }
+
+ @Override
+ public void onNotification(final Notification notification) {
+ getContext(notification.getImplementedInterface()).invoke(notification);
+ };
+
+ private InvokerContext getContext(final Class<?> type) {
+ return invokers.get(type);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Set<Class<? extends Notification>> getNotificationTypes(final Class<? extends org.opendaylight.yangtools.yang.binding.NotificationListener> type) {
+ // TODO: Investigate possibility and performance impact if we cache this or expose
+ // it from NotificationListenerInvoker
+ final Set<Class<? extends Notification>> ret = new HashSet<>();
+ for(final Method method : type.getMethods()) {
+ if(BindingReflections.isNotificationCallback(method)) {
+ final Class<? extends Notification> notification = (Class<? extends Notification>) method.getParameterTypes()[0];
+ ret.add(notification);
+ }
+ }
+ return ret;
+ }
+
+ private class InvokerContext {
+
+ private final QName name;
+ private final NotificationListenerInvoker invoker;
+
+ private InvokerContext(final QName name, final NotificationListenerInvoker invoker) {
+ this.name = name;
+ this.invoker = invoker;
+ }
+
+ public void invoke(final Notification notification) {
+ invoker.invokeNotification(delegate, name, notification);
+ }
+
+ }
+
+}
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.controller.sal.binding.impl;
+package org.opendaylight.controller.md.sal.binding.compat;
import org.opendaylight.controller.sal.binding.api.NotificationListener;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-package org.opendaylight.controller.sal.binding.impl;
+package org.opendaylight.controller.md.sal.binding.compat;
import org.opendaylight.yangtools.yang.binding.Notification;
import org.slf4j.Logger;
return invokers.keySet();
}
- private static Map<SchemaPath, NotificationListenerInvoker> createInvokerMapFor(final Class<? extends NotificationListener> implClz) {
+ public static Map<SchemaPath, NotificationListenerInvoker> createInvokerMapFor(final Class<? extends NotificationListener> implClz) {
final Map<SchemaPath, NotificationListenerInvoker> builder = new HashMap<>();
for(final Class<?> iface : implClz.getInterfaces()) {
if(NotificationListener.class.isAssignableFrom(iface) && BindingReflections.isBindingClass(iface)) {
import org.opendaylight.controller.md.sal.binding.impl.BindingDOMAdapterBuilder.Factory;
import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
import org.opendaylight.controller.md.sal.dom.api.DOMService;
-import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory;
import org.opendaylight.yangtools.binding.data.codec.api.BindingNormalizedNodeSerializer;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
private final BindingNormalizedNodeSerializer codec;
private final DOMNotificationService domNotifService;
- public BindingDOMNotificationServiceAdapter(final BindingNormalizedNodeSerializer codec, final DOMNotificationService domNotifService, final NotificationInvokerFactory notificationInvokerFactory) {
+ public BindingDOMNotificationServiceAdapter(final BindingNormalizedNodeSerializer codec, final DOMNotificationService domNotifService) {
this.codec = codec;
this.domNotifService = domNotifService;
}
protected NotificationService createInstance(final BindingToNormalizedNodeCodec codec,
final ClassToInstanceMap<DOMService> delegates) {
final DOMNotificationService domNotification = delegates.getInstance(DOMNotificationService.class);
- final NotificationInvokerFactory invokerFactory = SingletonHolder.INVOKER_FACTORY;
- return new BindingDOMNotificationServiceAdapter(codec.getCodecRegistry(), domNotification, invokerFactory);
+ return new BindingDOMNotificationServiceAdapter(codec.getCodecRegistry(), domNotification);
}
@Override
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen;
-
-import com.google.common.base.Preconditions;
-
-/**
- * Exception is raised when supplied Bidning Aware
- * RPCService class is not routed and was used in context
- * where routed RPCs should only be used.
- *
- */
-public class RpcIsNotRoutedException extends IllegalStateException {
-
- private static final long serialVersionUID = 1L;
-
- public RpcIsNotRoutedException(final String message, final Throwable cause) {
- super(Preconditions.checkNotNull(message), cause);
- }
-
- public RpcIsNotRoutedException(final String message) {
- super(Preconditions.checkNotNull(message));
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen;
-
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-
-public interface RuntimeCodeGenerator {
-
- /**
- * Returns an instance of provided RpcService type which delegates all calls
- * to the delegate.
- *
- * <p>
- * Returned instance:
- * <ul>
- * <li>implements provided subclass of RpcService type and
- * {@link org.opendaylight.controller.sal.binding.spi.DelegateProxy} interface.
- * <li>
- * <p>
- * delegates all invocations of methods, which are defined in RpcService
- * subtype to delegate which is defined by
- * {@link org.opendaylight.controller.sal.binding.spi.DelegateProxy#setDelegate(Object)}.
- * <p>
- * If delegate is not defined (<code>getDelegate() == null</code>)
- * implementation throws {@link IllegalStateException}
- * <li>{@link org.opendaylight.controller.sal.binding.spi.DelegateProxy#getDelegate()} - returns the delegate to which
- * all calls are delegated.
- * <li>{@link org.opendaylight.controller.sal.binding.spi.DelegateProxy#setDelegate(Object)} - sets the delegate for
- * particular instance
- *
- * </ul>
- *
- * @param serviceType
- * - Subclass of RpcService for which direct proxy is to be
- * generated.
- * @return Instance of RpcService of provided serviceType which implements
- * and {@link org.opendaylight.controller.sal.binding.spi.DelegateProxy}
- * @throws IllegalArgumentException
- *
- */
- <T extends RpcService> T getDirectProxyFor(Class<T> serviceType) throws IllegalArgumentException;
-
- /**
- * Returns an instance of provided RpcService type which routes all calls to
- * other instances selected on particular input field.
- *
- * <p>
- * Returned instance:
- * <ul>
- * <li>Implements:
- * <ul>
- * <li>{@link org.opendaylight.controller.sal.binding.spi.DelegateProxy}
- * <li>{@link RpcRouter}
- * </ul>
- * <li>
- * routes all invocations of methods, which are defined in RpcService
- * subtype based on method arguments and routing information defined in the
- * RpcRoutingTables for this instance
- * {@link RpcRouter#getRoutingTable(Class)}.
- * <ul>
- * <li>
- * Implementation uses
- * {@link RpcRouter#getService(Class, org.opendaylight.yangtools.yang.binding.InstanceIdentifier)} method to
- * retrieve particular instance to which call will be routed.
- * <li>
- * Instance of {@link org.opendaylight.yangtools.yang.binding.InstanceIdentifier} is determined by first argument of
- * method and is retrieved via method which is annotated with
- * {@link org.opendaylight.yangtools.yang.binding.annotations.RoutingContext}.
- * Class representing Routing Context Identifier is retrieved by a
- * {@link org.opendaylight.yangtools.yang.binding.annotations.RoutingContext}.
- * <li>If first argument is not defined / {@link org.opendaylight.yangtools.yang.binding.annotations.RoutingContext} annotation
- * is not present on any field invocation will be delegated to default
- * service {@link RpcRouter#getDefaultService()}.
- * </ul>
- *
- * @param serviceType
- * - Subclass of RpcService for which Router is to be generated.
- * @return Instance of RpcService of provided serviceType which implements
- * also {@link RpcRouter}<T> and {@link org.opendaylight.controller.sal.binding.spi.DelegateProxy}
- * @throws RpcIsNotRoutedException
- */
- <T extends RpcService> RpcRouter<T> getRouterFor(Class<T> serviceType,String name) throws IllegalArgumentException, RpcIsNotRoutedException;
-
- NotificationInvokerFactory getInvokerFactory();
-}
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen;
-
-import java.lang.reflect.Field;
-import java.util.Map;
-
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-
-public final class RuntimeCodeHelper {
- private RuntimeCodeHelper() {
- throw new UnsupportedOperationException("Utility class should never be instantiated");
- }
-
- private static Field getField(final Class<?> cls, final String name) {
- try {
- return cls.getField(name);
- } catch (NoSuchFieldException e) {
- throw new IllegalArgumentException(
- String.format("Class %s is missing field %s", cls, name), e);
- } catch (SecurityException e) {
- throw new IllegalStateException(String.format("Failed to examine class %s", cls), e);
- }
- }
-
- private static Field getDelegateField(final Class<?> cls) {
- return getField(cls, RuntimeCodeSpecification.DELEGATE_FIELD);
- }
-
- private static Object getFieldValue(final Field field, final Object obj) {
- try {
- return field.get(obj);
- } catch (IllegalAccessException e) {
- throw new IllegalStateException(String.format("Failed to get field %s of object %s", field, obj), e);
- }
- }
-
- private static void setFieldValue(final Field field, final Object obj, final Object value) {
- try {
- field.set(obj, value);
- } catch (IllegalAccessException e) {
- throw new IllegalStateException(String.format("Failed to set field %s to %s", field, value), e);
- }
- }
-
- /**
- * Helper method to return delegate from ManagedDirectedProxy with use of reflection.
- *
- * Note: This method uses reflection, but access to delegate field should be
- * avoided and called only if necessary.
- */
- @SuppressWarnings("unchecked")
- public static <T extends RpcService> T getDelegate(final RpcService proxy) {
- return (T)getFieldValue(getDelegateField(proxy.getClass()), proxy);
- }
-
- /**
- * Helper method to set delegate to ManagedDirectedProxy with use of reflection.
- *
- * Note: This method uses reflection, but setting delegate field should not occur too much
- * to introduce any significant performance hits.
- */
- public static void setDelegate(final Object proxy, final Object delegate) {
- final Field field = getDelegateField(proxy.getClass());
-
- if (delegate != null) {
- final Class<?> ft = field.getType();
- if (!ft.isAssignableFrom(delegate.getClass())) {
- throw new IllegalArgumentException(
- String.format("Field %s type %s is not compatible with delegate type %s",
- field, ft, delegate.getClass()));
- }
- }
-
- setFieldValue(field, proxy, delegate);
- }
-
- @SuppressWarnings("unchecked")
- public static Map<InstanceIdentifier<? extends Object>,? extends RpcService> getRoutingTable(final RpcService target, final Class<? extends BaseIdentity> tableClass) {
- final Field field = getField(target.getClass(), RuntimeCodeSpecification.getRoutingTableField(tableClass));
- return (Map<InstanceIdentifier<? extends Object>,? extends RpcService>) getFieldValue(field, target);
- }
-
- public static void setRoutingTable(final RpcService target, final Class<? extends BaseIdentity> tableClass, final Map<InstanceIdentifier<? extends Object>,? extends RpcService> routingTable) {
- final Field field = getField(target.getClass(), RuntimeCodeSpecification.getRoutingTableField(tableClass));
- setFieldValue(field, target, routingTable);
- }
-}
+++ /dev/null
-/**
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen;
-
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.NotificationListener;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-
-public final class RuntimeCodeSpecification {
- public final static String DIRECT_PROXY_SUFFIX = "DirectProxy";
- public final static String INVOKER_SUFFIX = "ListenerInvoker";
- public final static String ROUTER_SUFFIX = "Router";
-
- public final static String DELEGATE_FIELD = "_delegate";
- public final static String ROUTING_TABLE_FIELD_PREFIX = "_routes_";
-
- private RuntimeCodeSpecification() {
- throw new UnsupportedOperationException("Utility class");
- }
-
- /**
- * Returns a name for generated interface
- */
- private static String getGeneratedName(final Class<? extends Object> cls, final String suffix) {
- return cls.getName() + "$$Broker$" + suffix;
- }
-
- public static String getInvokerName(final Class<? extends NotificationListener> listener) {
- return getGeneratedName(listener, RuntimeCodeSpecification.INVOKER_SUFFIX);
- }
-
- /**
- * Returns a name for DirectProxy implementation
- */
- public static String getDirectProxyName(final Class<? extends RpcService> base) {
- return getGeneratedName(base, RuntimeCodeSpecification.DIRECT_PROXY_SUFFIX);
- }
-
- /**
- * Returns a name for Router implementation
- */
- public static String getRouterName(final Class<? extends RpcService> base) {
- return getGeneratedName(base, RuntimeCodeSpecification.ROUTER_SUFFIX);
- }
-
- /**
- * Returns a field name for specified routing context
- */
- public static String getRoutingTableField(final Class<? extends BaseIdentity> routingContext) {
- return "_routes_" + routingContext.getSimpleName();
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import com.google.common.base.Supplier;
-import com.google.common.collect.Iterables;
-import java.util.Map;
-import java.util.WeakHashMap;
-import javassist.ClassPool;
-import javassist.CtClass;
-import javassist.CtMethod;
-import javassist.NotFoundException;
-import javax.annotation.concurrent.GuardedBy;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
-import org.opendaylight.controller.sal.binding.codegen.RpcIsNotRoutedException;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory;
-import org.opendaylight.yangtools.sal.binding.generator.util.JavassistUtils;
-import org.opendaylight.yangtools.util.ClassLoaderUtils;
-import org.opendaylight.yangtools.yang.binding.BindingMapping;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.NotificationListener;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.opendaylight.yangtools.yang.binding.annotations.RoutingContext;
-
-abstract class AbstractRuntimeCodeGenerator implements org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator, NotificationInvokerFactory {
- @GuardedBy("this")
- private final Map<Class<? extends NotificationListener>, RuntimeGeneratedInvokerPrototype> invokerClasses = new WeakHashMap<>();
- private final CtClass brokerNotificationListener;
- protected final JavassistUtils utils;
-
- protected AbstractRuntimeCodeGenerator(final ClassPool pool) {
- utils = JavassistUtils.forClassPool(pool);
-
- /*
- * Make sure Javassist ClassPool sees the classloader of RpcService
- */
- utils.ensureClassLoader(RpcService.class);
-
- brokerNotificationListener = utils.asCtClass(org.opendaylight.controller.sal.binding.api.NotificationListener.class);
- }
-
- protected final CtClass getBrokerNotificationListener() {
- return brokerNotificationListener;
- }
-
- protected abstract RuntimeGeneratedInvokerPrototype generateListenerInvoker(Class<? extends NotificationListener> cls);
- protected abstract <T extends RpcService> Supplier<T> directProxySupplier(final Class<T> serviceType);
- protected abstract <T extends RpcService> Supplier<T> routerSupplier(final Class<T> serviceType, RpcServiceMetadata metadata);
-
- private RpcServiceMetadata getRpcMetadata(final CtClass iface) throws ClassNotFoundException, NotFoundException, RpcIsNotRoutedException {
- final RpcServiceMetadata metadata = new RpcServiceMetadata();
-
- for (CtMethod method : iface.getMethods()) {
- if (isRpcMethodWithInput(iface, method)) {
- final RpcMetadata routingPair = getRpcMetadata(method);
- if (routingPair != null) {
- metadata.addContext(routingPair.getContext());
- metadata.addRpcMethod(method.getName(), routingPair);
-
- /*
- * Force-load the RPC class representing the "input" of this RPC.
- *
- * FIXME: this is pre-existing side-effect of the original code, which
- * kept a reference to the loaded class, but it did not use it.
- *
- * There was no explanation as to why forcing this load was
- * necessary. As far as I can tell now is that it forces the
- * resolution of method arguments, which would (according to
- * my reading of JLS) occur only when the method is invoked via
- * binding-aware class action, not when coming from
- * binding-independent world. Whether that makes sense or not,
- * remains to be investigated.
- */
- Thread.currentThread().getContextClassLoader().loadClass(routingPair.getInputType().getName());
- } else {
- throw new RpcIsNotRoutedException(String.format("RPC %s from %s is not routed", method.getName(), iface.getName()));
- }
- }
- }
-
- return metadata;
- }
-
-
- private boolean isRpcMethodWithInput(final CtClass iface, final CtMethod method) throws NotFoundException {
- if(iface.equals(method.getDeclaringClass())
- && method.getParameterTypes().length == 1) {
- final CtClass onlyArg = method.getParameterTypes()[0];
- if(onlyArg.isInterface() && onlyArg.getName().endsWith(BindingMapping.RPC_INPUT_SUFFIX)) {
- return true;
- }
- }
- return false;
- }
-
- private RpcMetadata getRpcMetadata(final CtMethod method) throws NotFoundException {
- final CtClass inputClass = method.getParameterTypes()[0];
- return rpcMethodMetadata(inputClass, inputClass, method.getName());
- }
-
- private RpcMetadata rpcMethodMetadata(final CtClass dataClass, final CtClass inputClass, final String rpcMethod) throws NotFoundException {
- for (CtMethod method : dataClass.getMethods()) {
- if (method.getName().startsWith("get") && method.getParameterTypes().length == 0) {
- for (Object annotation : method.getAvailableAnnotations()) {
- if (annotation instanceof RoutingContext) {
- boolean encapsulated = !method.getReturnType().equals(utils.asCtClass(InstanceIdentifier.class));
- return new RpcMetadata(rpcMethod, ((RoutingContext)annotation).value(), method, encapsulated, inputClass);
- }
- }
- }
- }
-
- for (CtClass iface : dataClass.getInterfaces()) {
- final RpcMetadata ret = rpcMethodMetadata(iface, inputClass, rpcMethod);
- if(ret != null) {
- return ret;
- }
- }
- return null;
- }
-
- private synchronized RuntimeGeneratedInvokerPrototype resolveInvokerClass(final Class<? extends NotificationListener> cls) {
- RuntimeGeneratedInvokerPrototype invoker = invokerClasses.get(cls);
- if (invoker != null) {
- return invoker;
- }
-
- synchronized (utils) {
- invoker = ClassLoaderUtils.withClassLoader(cls.getClassLoader(), new Supplier<RuntimeGeneratedInvokerPrototype>() {
- @Override
- public RuntimeGeneratedInvokerPrototype get() {
- return generateListenerInvoker(cls);
- }
- });
- }
-
- invokerClasses.put(cls, invoker);
- return invoker;
- }
-
- @Override
- public final NotificationInvokerFactory getInvokerFactory() {
- return this;
- }
-
- @Override
- public final <T extends RpcService> T getDirectProxyFor(final Class<T> serviceType) {
- synchronized (utils) {
- return ClassLoaderUtils.withClassLoader(serviceType.getClassLoader(), directProxySupplier(serviceType));
- }
- }
-
- @Override
- public final <T extends RpcService> RpcRouter<T> getRouterFor(final Class<T> serviceType, final String name) throws RpcIsNotRoutedException {
- final RpcServiceMetadata metadata = ClassLoaderUtils.withClassLoader(serviceType.getClassLoader(), new Supplier<RpcServiceMetadata>() {
- @Override
- public RpcServiceMetadata get() {
- try {
- return getRpcMetadata(utils.asCtClass(serviceType));
- } catch (ClassNotFoundException | NotFoundException e) {
- throw new IllegalStateException(String.format("Failed to load metadata for class %s", serviceType), e);
- }
- }
- });
-
- if (Iterables.isEmpty(metadata.getContexts())) {
- throw new RpcIsNotRoutedException("Service doesn't have routing context associated.");
- }
-
- synchronized (utils) {
- final T instance = ClassLoaderUtils.withClassLoader(serviceType.getClassLoader(), routerSupplier(serviceType, metadata));
- return new RpcRouterCodegenInstance<T>(name, serviceType, instance, metadata.getContexts());
- }
- }
-
- @Override
- public NotificationInvoker invokerFor(final NotificationListener instance) {
- final Class<? extends NotificationListener> cls = instance.getClass();
- final RuntimeGeneratedInvokerPrototype prototype = resolveInvokerClass(cls);
-
- try {
- return RuntimeGeneratedInvoker.create(instance, prototype);
- } catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException(String.format("Failed to create invoker for %s", instance), e);
- }
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import com.google.common.base.Supplier;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.ImmutableSet.Builder;
-import java.lang.reflect.Method;
-import java.util.Map;
-import javassist.CannotCompileException;
-import javassist.ClassPool;
-import javassist.CtClass;
-import javassist.CtMethod;
-import javassist.NotFoundException;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeSpecification;
-import org.opendaylight.yangtools.sal.binding.generator.util.ClassGenerator;
-import org.opendaylight.yangtools.sal.binding.generator.util.MethodGenerator;
-import org.opendaylight.yangtools.util.ClassLoaderUtils;
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.Notification;
-import org.opendaylight.yangtools.yang.binding.NotificationListener;
-import org.opendaylight.yangtools.yang.binding.RpcImplementation;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.opendaylight.yangtools.yang.binding.util.BindingReflections;
-
-final class DefaultRuntimeCodeGenerator extends AbstractRuntimeCodeGenerator {
-
- DefaultRuntimeCodeGenerator(final ClassPool pool) {
- super(pool);
- }
-
- @Override
- protected <T extends RpcService> Supplier<T> directProxySupplier(final Class<T> serviceType) {
- return new Supplier<T>() {
- @SuppressWarnings("unchecked")
- @Override
- public T get() {
- final String proxyName = RuntimeCodeSpecification.getDirectProxyName(serviceType);
-
- final Class<?> potentialClass = ClassLoaderUtils.tryToLoadClassWithTCCL(proxyName);
- if (potentialClass != null) {
- try {
- return (T)potentialClass.newInstance();
- } catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException("Failed to instantiate class " + potentialClass.getName(), e);
- }
- }
-
- final CtClass supertype = utils.asCtClass(serviceType);
- final String directProxyName = RuntimeCodeSpecification.getDirectProxyName(serviceType);
-
- final CtClass createdCls;
- try {
- createdCls = utils.createClass(directProxyName, supertype, new ClassGenerator() {
- @Override
- public void process(final CtClass cls) throws CannotCompileException {
- utils.field(cls, RuntimeCodeSpecification.DELEGATE_FIELD, serviceType);
- utils.implementsType(cls, utils.asCtClass(RpcImplementation.class));
- utils.implementMethodsFrom(cls, supertype, new MethodGenerator() {
- @Override
- public void process(final CtMethod method) throws CannotCompileException {
- final StringBuilder sb = new StringBuilder("\n");
- sb.append("{\n");
- sb.append(" if (").append(RuntimeCodeSpecification.DELEGATE_FIELD).append(" == null) {\n");
- sb.append(" throw new java.lang.IllegalStateException(\"No default provider is available\");\n");
- sb.append(" }\n");
- sb.append(" return ($r) ").append(RuntimeCodeSpecification.DELEGATE_FIELD).append('.').append(method.getName()).append("($$);\n");
- sb.append("}\n");
- method.setBody(sb.toString());
- }
- });
-
- // FIXME: copy this one...
- utils.implementMethodsFrom(cls, utils.asCtClass(RpcImplementation.class), new MethodGenerator() {
- @Override
- public void process(final CtMethod method) throws CannotCompileException {
- final StringBuilder sb = new StringBuilder("\n");
- sb.append("{\n");
- sb.append(" throw new java.lang.IllegalStateException(\"No provider is processing supplied message\");\n");
- sb.append(" return ($r) null;\n");
- sb.append("}\n");
- method.setBody(sb.toString());
- }
- });
- }
- });
- } catch (CannotCompileException e) {
- throw new IllegalStateException("Failed to create class " + directProxyName, e);
- }
-
- final Class<?> c;
- try {
- c = createdCls.toClass(serviceType.getClassLoader(), serviceType.getProtectionDomain());
- } catch (CannotCompileException e) {
- throw new IllegalStateException(String.format("Failed to create class %s", createdCls), e);
- }
-
- try {
- return (T) c.newInstance();
- } catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException(String.format("Failed to instantiated class %s", c), e);
- }
- }
- };
- }
-
- @Override
- protected <T extends RpcService> Supplier<T> routerSupplier(final Class<T> serviceType, final RpcServiceMetadata metadata) {
- return new Supplier<T>() {
- @SuppressWarnings("unchecked")
- @Override
- public T get() {
- final CtClass supertype = utils.asCtClass(serviceType);
- final String routerName = RuntimeCodeSpecification.getRouterName(serviceType);
- final Class<?> potentialClass = ClassLoaderUtils.tryToLoadClassWithTCCL(routerName);
- if (potentialClass != null) {
- try {
- return (T)potentialClass.newInstance();
- } catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException("Failed to instantiate class", e);
- }
- }
-
- final CtClass targetCls;
- try {
- targetCls = utils.createClass(routerName, supertype, new ClassGenerator() {
- @Override
- public void process(final CtClass cls) throws CannotCompileException {
- utils.field(cls, RuntimeCodeSpecification.DELEGATE_FIELD, serviceType);
- //utils.field(cls, REMOTE_INVOKER_FIELD,iface);
- utils.implementsType(cls, utils.asCtClass(RpcImplementation.class));
-
- for (final Class<? extends BaseIdentity> ctx : metadata.getContexts()) {
- utils.field(cls, RuntimeCodeSpecification.getRoutingTableField(ctx), Map.class);
- }
-
- utils.implementMethodsFrom(cls, supertype, new MethodGenerator() {
- @Override
- public void process(final CtMethod method) throws CannotCompileException {
- final int ptl;
- try {
- ptl = method.getParameterTypes().length;
- } catch (NotFoundException e) {
- throw new CannotCompileException(e);
- }
- final StringBuilder sb = new StringBuilder();
-
- switch (ptl) {
- case 0:
- sb.append("return ($r) ").append(RuntimeCodeSpecification.DELEGATE_FIELD).append('.').append(method.getName()).append("($$);");
- break;
- case 1:
- final RpcMetadata rpcMeta = metadata.getRpcMethod(method.getName());
- final String rtGetter = rpcMeta.getInputRouteGetter().getName();
- final String stName = supertype.getName();
-
- sb.append('\n');
- sb.append("{\n");
- sb.append(" if ($1 == null) {\n");
- sb.append(" throw new IllegalArgumentException(\"RPC input must not be null and must contain a value for field ").append(rtGetter).append("\");\n");
- sb.append(" }\n");
- sb.append(" if ($1.").append(rtGetter).append("() == null) {\n");
- sb.append(" throw new IllegalArgumentException(\"Field ").append(rtGetter).append(" must not be null\");\n");
- sb.append(" }\n");
-
- sb.append(" final org.opendaylight.yangtools.yang.binding.InstanceIdentifier identifier = $1.").append(rtGetter).append("()");
- if (rpcMeta.isRouteEncapsulated()) {
- sb.append(".getValue()");
- }
- sb.append(";\n");
-
- sb.append(" ").append(supertype.getName()).append(" instance = (").append(stName).append(") ").append(RuntimeCodeSpecification.getRoutingTableField(rpcMeta.getContext())).append(".get(identifier);\n");
- sb.append(" if (instance == null) {\n");
- sb.append(" instance = ").append(RuntimeCodeSpecification.DELEGATE_FIELD).append(";\n");
- sb.append(" }\n");
-
- sb.append(" if (instance == null) {\n");
- sb.append(" throw new java.lang.IllegalStateException(\"No routable provider is processing routed message for \" + String.valueOf(identifier));\n");
- sb.append(" }\n");
- sb.append(" return ($r) instance.").append(method.getName()).append("($$);\n");
- sb.append('}');
- break;
- default:
- throw new CannotCompileException(String.format("Unsupported parameters length %s", ptl));
- }
-
- method.setBody(sb.toString());
- }
- });
-
- // FIXME: move this into a template class
- utils.implementMethodsFrom(cls, utils.asCtClass(RpcImplementation.class), new MethodGenerator() {
- @Override
- public void process(final CtMethod method) throws CannotCompileException {
- final StringBuilder sb = new StringBuilder("\n");
- sb.append("{\n");
- sb.append(" throw new java.lang.IllegalStateException(\"No provider is processing supplied message\");\n");
- sb.append(" return ($r) null;\n");
- sb.append("}\n");
-
- method.setBody(sb.toString());
- }
- });
- }
- });
- } catch (CannotCompileException e) {
- throw new IllegalStateException("Failed to create class " + routerName, e);
- }
-
- final Class<?> c;
- try {
- c = targetCls.toClass(serviceType.getClassLoader(), serviceType.getProtectionDomain());
- } catch (CannotCompileException e) {
- throw new IllegalStateException(String.format("Failed to compile class %s", targetCls), e);
- }
-
- try {
- return (T)c.newInstance();
- } catch (InstantiationException | IllegalAccessException e) {
- throw new IllegalStateException(String.format("Failed to instantiate class %s", c), e);
- }
- }
- };
- }
-
- @SuppressWarnings("unchecked")
- @Override
- protected RuntimeGeneratedInvokerPrototype generateListenerInvoker(final Class<? extends NotificationListener> listenerType) {
- final String invokerName = RuntimeCodeSpecification.getInvokerName(listenerType);
- final CtClass targetCls;
-
- // Builder for a set of supported types. Filled while the target class is being generated
- final Builder<Class<? extends Notification>> b = ImmutableSet.builder();
-
- try {
- targetCls = utils.createClass(invokerName, getBrokerNotificationListener(), new ClassGenerator() {
- @Override
- public void process(final CtClass cls) throws CannotCompileException {
- utils.field(cls, RuntimeCodeSpecification.DELEGATE_FIELD, listenerType);
- utils.implementMethodsFrom(cls, getBrokerNotificationListener(), new MethodGenerator() {
- @Override
- public void process(final CtMethod method) throws CannotCompileException {
- final StringBuilder sb = new StringBuilder("\n");
-
- sb.append("{\n");
-
- for (Method m : listenerType.getMethods()) {
- if (BindingReflections.isNotificationCallback(m)) {
- final Class<?> argType = m.getParameterTypes()[0];
-
- // populates builder above
- b.add((Class<? extends Notification>) argType);
-
- sb.append(" if ($1 instanceof ").append(argType.getName()).append(") {\n");
- sb.append(" ").append(RuntimeCodeSpecification.DELEGATE_FIELD).append('.').append(m.getName()).append("((").append(argType.getName()).append(") $1);\n");
- sb.append(" return null;\n");
- sb.append(" } else ");
- }
- }
-
- sb.append(" return null;\n");
- sb.append("}\n");
- method.setBody(sb.toString());
- }
- });
- }
- });
- } catch (CannotCompileException e) {
- throw new IllegalStateException("Failed to create class " + invokerName, e);
- }
-
- final Class<?> finalClass;
- try {
- finalClass = targetCls.toClass(listenerType.getClassLoader(), listenerType.getProtectionDomain());
- } catch (CannotCompileException e) {
- throw new IllegalStateException(String.format("Failed to compile class %s", targetCls), e);
- }
-
- return new RuntimeGeneratedInvokerPrototype(b.build(), (Class<? extends org.opendaylight.controller.sal.binding.api.NotificationListener<?>>) finalClass);
- }
-}
+++ /dev/null
-/**
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import javassist.CtClass;
-import javassist.CtMethod;
-
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-
-final class RpcMetadata {
- private final Class<? extends BaseIdentity> context;
- private final CtMethod inputRouteGetter;
- private final Boolean routeEncapsulated;
- private final CtClass inputType;
- private final String methodName;
-
- public Class<? extends BaseIdentity> getContext() {
- return context;
- }
-
- public CtMethod getInputRouteGetter() {
- return inputRouteGetter;
- }
-
- public CtClass getInputType() {
- return inputType;
- }
-
- public boolean isRouteEncapsulated() {
- return routeEncapsulated;
- }
-
- public RpcMetadata(final String methodName, final Class<? extends BaseIdentity> context, final CtMethod inputRouteGetter, final boolean routeEncapsulated, final CtClass inputType) {
- this.inputRouteGetter = Preconditions.checkNotNull(inputRouteGetter);
- this.methodName = Preconditions.checkNotNull(methodName);
- this.inputType = Preconditions.checkNotNull(inputType);
- this.context = Preconditions.checkNotNull(context);
- this.routeEncapsulated = routeEncapsulated;
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + methodName.hashCode();
- result = prime * result + context.hashCode();
- result = prime * result + inputRouteGetter.hashCode();
- result = prime * result + routeEncapsulated.hashCode();
- result = prime * result + inputType.hashCode();
- return result;
- }
-
- @Override
- public boolean equals(final Object obj) {
- if (this == obj) {
- return true;
- }
- if (!(obj instanceof RpcMetadata)) {
- return false;
- }
- final RpcMetadata other = (RpcMetadata) obj;
- if (!methodName.equals(other.methodName)) {
- return false;
- }
- if (!context.equals(other.context)) {
- return false;
- }
- if (!inputRouteGetter.equals(other.inputRouteGetter)) {
- return false;
- }
- if (!routeEncapsulated.equals(other.routeEncapsulated)) {
- return false;
- }
- return inputType.equals(other.inputType);
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("context", context)
- .add("inputRouteGetter", inputRouteGetter)
- .add("inputType", inputType)
- .add("methodName", methodName)
- .add("routeEncapsulated", routeEncapsulated)
- .toString();
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import static org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper.setRoutingTable;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import javax.annotation.concurrent.GuardedBy;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRoutingTable;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper;
-import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.ListenerRegistry;
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RpcRouterCodegenInstance<T extends RpcService> implements //
-RpcRouter<T>, RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>> {
-
- private static final Logger LOG = LoggerFactory.getLogger(RpcRouterCodegenInstance.class);
-
- private final Class<T> serviceType;
-
- private final T invocationProxy;
-
- private final Set<Class<? extends BaseIdentity>> contexts;
-
- private final ListenerRegistry<RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>>> listeners;
-
- private final Map<Class<? extends BaseIdentity>, RpcRoutingTableImpl<? extends BaseIdentity, T>> routingTables;
-
- @SuppressWarnings("unchecked")
- public RpcRouterCodegenInstance(final String name,final Class<T> type, final T routerImpl, final Iterable<Class<? extends BaseIdentity>> contexts) {
- this.listeners = ListenerRegistry.create();
- this.serviceType = type;
- this.invocationProxy = routerImpl;
- this.contexts = ImmutableSet.copyOf(contexts);
- Map<Class<? extends BaseIdentity>, RpcRoutingTableImpl<? extends BaseIdentity, T>> mutableRoutingTables = new HashMap<>();
- for (Class<? extends BaseIdentity> ctx : contexts) {
- RpcRoutingTableImpl<? extends BaseIdentity, T> table = new RpcRoutingTableImpl<>(name,ctx,type);
-
- @SuppressWarnings("rawtypes")
- Map invokerView = table.getRoutes();
-
- setRoutingTable(invocationProxy, ctx, invokerView);
- mutableRoutingTables.put(ctx, table);
- table.registerRouteChangeListener(this);
- }
- this.routingTables = ImmutableMap.copyOf(mutableRoutingTables);
- }
-
- @Override
- public Class<T> getServiceType() {
- return serviceType;
- }
-
- @Override
- public T getInvocationProxy() {
- return invocationProxy;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public <C extends BaseIdentity> RpcRoutingTable<C, T> getRoutingTable(final Class<C> routeContext) {
- return (RpcRoutingTable<C, T>) routingTables.get(routeContext);
- }
-
- @Override
- public T getDefaultService() {
- return RuntimeCodeHelper.getDelegate(invocationProxy);
- }
-
- @Override
- public Set<Class<? extends BaseIdentity>> getContexts() {
- return contexts;
- }
-
- @Override
- public <L extends RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>>> ListenerRegistration<L> registerRouteChangeListener(
- final L listener) {
- return listeners.registerWithType(listener);
- }
-
- @Override
- public void onRouteChange(final RouteChange<Class<? extends BaseIdentity>, InstanceIdentifier<?>> change) {
- for (ListenerRegistration<RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>>> listener : listeners) {
- try {
- listener.getInstance().onRouteChange(change);
- } catch (Exception e) {
- LOG.error("Error occured during invoker listener {}", listener.getInstance(), e);
- }
- }
- }
-
- @Override
- public T getService(final Class<? extends BaseIdentity> context, final InstanceIdentifier<?> path) {
- return routingTables.get(context).getRoute(path);
- }
-
- @Override
- public RoutedRpcRegistration<T> addRoutedRpcImplementation(final T service) {
- return new RoutedRpcRegistrationImpl(service);
- }
-
- public void removeDefaultImplementation(final T instance) {
- RpcService current = RuntimeCodeHelper.getDelegate(invocationProxy);
- if(instance == current) {
- RuntimeCodeHelper.setDelegate(invocationProxy, null);
- }
- }
-
- @Override
- public RpcRegistration<T> registerDefaultService(final T service) {
- RuntimeCodeHelper.setDelegate(invocationProxy, service);
- return new DefaultRpcImplementationRegistration(service);
- }
-
- private final class RoutedRpcRegistrationImpl extends AbstractObjectRegistration<T> implements RoutedRpcRegistration<T> {
- /*
- * FIXME: retaining this collection is not completely efficient. We really should be storing
- * a reference to this registration, as a particular listener may be registered multiple
- * times -- and then this goes kaboom in various aspects.
- */
- @GuardedBy("this")
- private final Collection<Class<? extends BaseIdentity>> contexts = new ArrayList<>(1);
-
- public RoutedRpcRegistrationImpl(final T instance) {
- super(instance);
- }
-
- @Override
- public Class<T> getServiceType() {
- return serviceType;
- }
-
- @Override
- public synchronized void registerPath(final Class<? extends BaseIdentity> context, final InstanceIdentifier<?> path) {
- if (isClosed()) {
- LOG.debug("Closed registration of {} ignoring new path {}", getInstance(), path);
- return;
- }
-
- routingTables.get(context).updateRoute(path, getInstance());
- contexts.add(context);
- }
-
- @Override
- public synchronized void unregisterPath(final Class<? extends BaseIdentity> context, final InstanceIdentifier<?> path) {
- if (isClosed()) {
- LOG.debug("Closed unregistration of {} ignoring new path {}", getInstance(), path);
- return;
- }
-
- routingTables.get(context).removeRoute(path, getInstance());
- contexts.remove(context);
- }
-
- @Deprecated
- @Override
- public void registerInstance(final Class<? extends BaseIdentity> context, final InstanceIdentifier<?> instance) {
- registerPath(context, instance);
- }
-
- @Deprecated
- @Override
- public void unregisterInstance(final Class<? extends BaseIdentity> context, final InstanceIdentifier<?> instance) {
- unregisterPath(context, instance);
- }
-
- @Override
- protected synchronized void removeRegistration() {
- for (Class<? extends BaseIdentity> ctx : contexts) {
- routingTables.get(ctx).removeAllReferences(getInstance());
- }
- contexts.clear();
- }
- }
-
- private final class DefaultRpcImplementationRegistration extends AbstractObjectRegistration<T> implements RpcRegistration<T> {
-
-
- protected DefaultRpcImplementationRegistration(final T instance) {
- super(instance);
- }
-
- @Override
- protected void removeRegistration() {
- removeDefaultImplementation(this.getInstance());
- }
-
- @Override
- public Class<T> getServiceType() {
- return serviceType;
- }
- }
-
-
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
-import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRoutingTable;
-import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.concepts.Mutable;
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-final class RpcRoutingTableImpl<C extends BaseIdentity, S extends RpcService> implements
- Mutable, //
- RpcRoutingTable<C, S>, //
- RouteChangePublisher<Class<? extends BaseIdentity>, InstanceIdentifier<?>> {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(RpcRoutingTableImpl.class);
- private final String routerName;
- private final Class<S> serviceType;
-
- private final Class<C> contextType;
- private final ConcurrentMap<InstanceIdentifier<?>, S> routes;
- private final Map<InstanceIdentifier<?>, S> unmodifiableRoutes;
-
- private RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>> listener;
- private S defaultRoute;
-
- public RpcRoutingTableImpl(final String routerName,final Class<C> contextType, final Class<S> serviceType) {
- super();
- this.routerName = routerName;
- this.serviceType = serviceType;
- this.contextType = contextType;
- this.routes = new ConcurrentHashMap<>();
- this.unmodifiableRoutes = Collections.unmodifiableMap(routes);
- }
-
- @Override
- public void setDefaultRoute(final S target) {
- defaultRoute = target;
- }
-
- @Override
- public S getDefaultRoute() {
- return defaultRoute;
- }
-
- @Override
- public <L extends RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>>> ListenerRegistration<L> registerRouteChangeListener(
- final L listener) {
- return new SingletonListenerRegistration<L>(listener);
- }
-
- @Override
- public Class<C> getIdentifier() {
- return contextType;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public void updateRoute(final InstanceIdentifier<?> path, final S service) {
- S previous = this.routes.put(path, service);
-
- LOGGER.debug("Route {} updated to {} in routing table {}",path,service,this);
- @SuppressWarnings("rawtypes")
- RouteChangeListener listenerCapture = listener;
- if (previous == null && listenerCapture != null) {
- listenerCapture.onRouteChange(RoutingUtils.announcementChange(contextType, path));
- }
- }
-
-
- @Override
- @SuppressWarnings("unchecked")
- public void removeRoute(final InstanceIdentifier<?> path) {
- S previous = this.routes.remove(path);
- LOGGER.debug("Route {} to {} removed in routing table {}",path,previous,this);
- @SuppressWarnings("rawtypes")
- RouteChangeListener listenerCapture = listener;
- if (previous != null && listenerCapture != null) {
- listenerCapture.onRouteChange(RoutingUtils.removalChange(contextType, path));
- }
- }
-
- void removeRoute(final InstanceIdentifier<?> path, final S service) {
- @SuppressWarnings("rawtypes")
- RouteChangeListener listenerCapture = listener;
- if (routes.remove(path, service) && listenerCapture != null) {
- LOGGER.debug("Route {} to {} removed in routing table {}",path,service,this);
- listenerCapture.onRouteChange(RoutingUtils.removalChange(contextType, path));
- }
- }
-
- @Override
- public S getRoute(final InstanceIdentifier<?> nodeInstance) {
- S route = routes.get(nodeInstance);
- if (route != null) {
- return route;
- }
- return getDefaultRoute();
- }
-
- @Override
- public Map<InstanceIdentifier<?>, S> getRoutes() {
- return unmodifiableRoutes;
- }
-
- void removeAllReferences(final S service) {
- // FIXME: replace this via properly-synchronized BiMap (or something)
- final Iterator<S> it = routes.values().iterator();
- while (it.hasNext()) {
- final S s = it.next();
- if (service.equals(s)) {
- it.remove();
- }
- }
- }
-
- @Override
- public String toString() {
- return "RpcRoutingTableImpl [router=" + routerName + ", service=" + serviceType.getSimpleName() + ", context="
- + contextType.getSimpleName() + "]";
- }
-
- private class SingletonListenerRegistration<L extends RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>>> extends
- AbstractObjectRegistration<L>
- implements ListenerRegistration<L> {
-
- public SingletonListenerRegistration(final L instance) {
- super(instance);
- listener = instance;
- }
-
- @Override
- protected void removeRegistration() {
- listener = null;
- }
- }
-}
+++ /dev/null
-/**
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-
-import com.google.common.collect.Iterables;
-
-final class RpcServiceMetadata {
- private final Set<Class<? extends BaseIdentity>> contexts = new HashSet<>();
- private final Map<String, RpcMetadata> rpcMethods = new HashMap<>();
- private final Iterable<Class<? extends BaseIdentity>> roContexts = Iterables.unmodifiableIterable(contexts);
-
- public Iterable<Class<? extends BaseIdentity>> getContexts() {
- return roContexts;
- }
-
- public RpcMetadata getRpcMethod(final String name) {
- return rpcMethods.get(name);
- }
-
- public void addContext(final Class<? extends BaseIdentity> context) {
- contexts.add(context);
- }
-
- public void addRpcMethod(final String name, final RpcMetadata routingPair) {
- rpcMethods.put(name, routingPair);
- }
-}
+++ /dev/null
-/**
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-import java.util.Set;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker;
-import org.opendaylight.yangtools.yang.binding.Notification;
-import org.opendaylight.yangtools.yang.binding.NotificationListener;
-
-final class RuntimeGeneratedInvoker implements NotificationInvoker {
- private final org.opendaylight.controller.sal.binding.api.NotificationListener<Notification> invocationProxy;
- private final RuntimeGeneratedInvokerPrototype prototype;
- private final NotificationListener delegate;
-
- @SuppressWarnings("unchecked")
- private RuntimeGeneratedInvoker(final NotificationListener delegate, final RuntimeGeneratedInvokerPrototype prototype, final org.opendaylight.controller.sal.binding.api.NotificationListener<?> proxy) {
- this.invocationProxy = (org.opendaylight.controller.sal.binding.api.NotificationListener<Notification>) proxy;
- this.delegate = Preconditions.checkNotNull(delegate);
- this.prototype = prototype;
- }
-
- public static RuntimeGeneratedInvoker create(final NotificationListener delegate, final RuntimeGeneratedInvokerPrototype prototype) throws InstantiationException, IllegalAccessException {
- final org.opendaylight.controller.sal.binding.api.NotificationListener<?> proxy = Preconditions.checkNotNull(prototype.getProtoClass().newInstance());
- RuntimeCodeHelper.setDelegate(proxy, delegate);
- return new RuntimeGeneratedInvoker(delegate, prototype, proxy);
- }
-
- @Override
- public NotificationListener getDelegate() {
- return delegate;
- }
-
- @Override
- public org.opendaylight.controller.sal.binding.api.NotificationListener<Notification> getInvocationProxy() {
- return invocationProxy;
- }
-
- @Override
- public Set<Class<? extends Notification>> getSupportedNotifications() {
- return prototype.getSupportedNotifications();
- }
-
- @Override
- public void close() {
- // Nothing to do
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + delegate.hashCode();
- result = prime * result + invocationProxy.hashCode();
- result = prime * result + prototype.hashCode();
- return result;
- }
-
- @Override
- public boolean equals(final Object obj) {
- if (this == obj) {
- return true;
- }
- if (!(obj instanceof RuntimeGeneratedInvoker)) {
- return false;
- }
- final RuntimeGeneratedInvoker other = (RuntimeGeneratedInvoker) obj;
- if (!delegate.equals(other.delegate)) {
- return false;
- }
- if (!invocationProxy.equals(other.invocationProxy)) {
- return false;
- }
- return prototype.equals(other.prototype);
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this).toString();
- }
-}
+++ /dev/null
-/**
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import java.util.Set;
-
-import org.opendaylight.controller.sal.binding.api.NotificationListener;
-import org.opendaylight.yangtools.yang.binding.Notification;
-
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-
-final class RuntimeGeneratedInvokerPrototype {
- private final Set<Class<? extends Notification>> supportedNotifications;
- private final Class<? extends NotificationListener<?>> protoClass;
-
- public RuntimeGeneratedInvokerPrototype(final Set<Class<? extends Notification>> supportedNotifications, final Class<? extends NotificationListener<?>> protoClass) {
- this.supportedNotifications = Preconditions.checkNotNull(supportedNotifications);
- this.protoClass = Preconditions.checkNotNull(protoClass);
- }
-
- public Set<Class<? extends Notification>> getSupportedNotifications() {
- return supportedNotifications;
- }
-
- public Class<? extends NotificationListener<?>> getProtoClass() {
- return protoClass;
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + supportedNotifications.hashCode();
- result = prime * result + protoClass.hashCode();
- return result;
- }
-
- @Override
- public boolean equals(final Object obj) {
- if (this == obj) {
- return true;
- }
- if (!(obj instanceof RuntimeGeneratedInvokerPrototype)) {
- return false;
- }
- final RuntimeGeneratedInvokerPrototype other = (RuntimeGeneratedInvokerPrototype) obj;
- if (!protoClass.equals(other.protoClass)) {
- return false;
- }
- return supportedNotifications.equals(other.supportedNotifications);
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("protoClass", protoClass)
- .add("supportedNotifications", supportedNotifications)
- .toString();
- }
-}
import java.util.concurrent.TimeUnit;
import javassist.ClassPool;
import org.apache.commons.lang3.StringUtils;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory;
import org.opendaylight.yangtools.sal.binding.generator.util.JavassistUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public static final ClassPool CLASS_POOL = ClassPool.getDefault();
public static final JavassistUtils JAVASSIST = JavassistUtils.forClassPool(CLASS_POOL);
- public static final org.opendaylight.controller.sal.binding.codegen.impl.DefaultRuntimeCodeGenerator RPC_GENERATOR_IMPL = new org.opendaylight.controller.sal.binding.codegen.impl.DefaultRuntimeCodeGenerator(
- CLASS_POOL);
- public static final RuntimeCodeGenerator RPC_GENERATOR = RPC_GENERATOR_IMPL;
- public static final NotificationInvokerFactory INVOKER_FACTORY = RPC_GENERATOR_IMPL.getInvokerFactory();
public static final int CORE_NOTIFICATION_THREADS = 4;
public static final int MAX_NOTIFICATION_THREADS = 32;
if (NOTIFICATION_EXECUTOR == null) {
int queueSize = MAX_NOTIFICATION_QUEUE_SIZE;
- String queueValue = System.getProperty(NOTIFICATION_QUEUE_SIZE_PROPERTY);
+ final String queueValue = System.getProperty(NOTIFICATION_QUEUE_SIZE_PROPERTY);
if (StringUtils.isNotBlank(queueValue)) {
try {
queueSize = Integer.parseInt(queueValue);
logger.trace("Queue size was set to {}", queueSize);
- } catch (NumberFormatException e) {
+ } catch (final NumberFormatException e) {
logger.warn("Cannot parse {} as set by {}, using default {}", queueValue,
NOTIFICATION_QUEUE_SIZE_PROPERTY, queueSize);
}
public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
try {
executor.getQueue().put(r);
- } catch (InterruptedException e) {
+ } catch (final InterruptedException e) {
throw new RejectedExecutionException("Interrupted while waiting on the queue", e);
}
}
@Deprecated
public static synchronized ListeningExecutorService getDefaultCommitExecutor() {
if (COMMIT_EXECUTOR == null) {
- ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-commit-%d").build();
+ final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-commit-%d").build();
/*
* FIXME: this used to be newCacheThreadPool(), but MD-SAL does not have transaction
* ordering guarantees, which means that using a concurrent threadpool results
* in inconsistent data being present. Once proper primitives are introduced,
* concurrency can be reintroduced.
*/
- ExecutorService executor = Executors.newSingleThreadExecutor(factory);
+ final ExecutorService executor = Executors.newSingleThreadExecutor(factory);
COMMIT_EXECUTOR = MoreExecutors.listeningDecorator(executor);
}
public static ExecutorService getDefaultChangeEventExecutor() {
if (CHANGE_EVENT_EXECUTOR == null) {
- ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-change-%d").build();
+ final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-change-%d").build();
/*
* FIXME: this used to be newCacheThreadPool(), but MD-SAL does not have transaction
* ordering guarantees, which means that using a concurrent threadpool results
* in inconsistent data being present. Once proper primitives are introduced,
* concurrency can be reintroduced.
*/
- ExecutorService executor = Executors.newSingleThreadExecutor(factory);
+ final ExecutorService executor = Executors.newSingleThreadExecutor(factory);
CHANGE_EVENT_EXECUTOR = MoreExecutors.listeningDecorator(executor);
}
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.impl;
-
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.opendaylight.controller.md.sal.common.api.data.DataCommitHandler;
-import org.opendaylight.controller.md.sal.common.api.data.DataReader;
-import org.opendaylight.controller.md.sal.common.impl.routing.AbstractDataReadRouter;
-import org.opendaylight.controller.md.sal.common.impl.service.AbstractDataBroker;
-import org.opendaylight.controller.sal.binding.api.data.DataChangeListener;
-import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
-import org.opendaylight.yangtools.concepts.Registration;
-import org.opendaylight.yangtools.yang.binding.DataObject;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.util.DataObjectReadingUtil;
-
-import com.google.common.base.Predicate;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableMap.Builder;
-import com.google.common.collect.Maps;
-
-@Deprecated
-public class DataBrokerImpl extends
- AbstractDataBroker<InstanceIdentifier<? extends DataObject>, DataObject, DataChangeListener> //
- implements DataProviderService, AutoCloseable {
-
- private final static class ContainsWildcarded implements Predicate<InstanceIdentifier<? extends DataObject>> {
-
- private final InstanceIdentifier<? extends DataObject> key;
-
- public ContainsWildcarded(final InstanceIdentifier<? extends DataObject> key) {
- this.key = key;
- }
-
- @Override
- public boolean apply(final InstanceIdentifier<? extends DataObject> input) {
- return key.containsWildcarded(input);
- }
- }
-
- private final static class IsContainedWildcarded implements Predicate<InstanceIdentifier<? extends DataObject>> {
-
- private final InstanceIdentifier<? extends DataObject> key;
-
- public IsContainedWildcarded(final InstanceIdentifier<? extends DataObject> key) {
- this.key = key;
- }
-
- @Override
- public boolean apply(final InstanceIdentifier<? extends DataObject> input) {
- return input.containsWildcarded(key);
- }
- }
-
- private final AtomicLong nextTransaction = new AtomicLong();
- private final AtomicLong createdTransactionsCount = new AtomicLong();
- private final DelegatingDataReadRouter router = new DelegatingDataReadRouter();
- private DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject> rootCommitHandler;
-
- public DataBrokerImpl() {
- setDataReadRouter(router);
- }
-
- public void setDataReadDelegate(final DataReader<InstanceIdentifier<? extends DataObject>, DataObject> delegate) {
- router.setDelegate(delegate);
- }
-
- public AtomicLong getCreatedTransactionsCount() {
- return createdTransactionsCount;
- }
-
- @Override
- public DataTransactionImpl beginTransaction() {
- String transactionId = "BA-" + nextTransaction.getAndIncrement();
- createdTransactionsCount.getAndIncrement();
- return new DataTransactionImpl(transactionId, this);
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- protected Predicate<InstanceIdentifier<? extends DataObject>> createContainsPredicate(
- final InstanceIdentifier<? extends DataObject> key) {
- return new ContainsWildcarded(key);
- }
-
- @Override
- protected Predicate<InstanceIdentifier<? extends DataObject>> createIsContainedPredicate(
- final InstanceIdentifier<? extends DataObject> key) {
- return new IsContainedWildcarded(key);
- }
-
- @SuppressWarnings({ "unchecked", "rawtypes" })
- @Override
- protected Map<InstanceIdentifier<? extends DataObject>, DataObject> deepGetBySubpath(
- final Map<InstanceIdentifier<? extends DataObject>, DataObject> dataSet,
- final InstanceIdentifier<? extends DataObject> path) {
- Builder<InstanceIdentifier<? extends DataObject>, DataObject> builder = ImmutableMap.builder();
- Map<InstanceIdentifier<? extends DataObject>, DataObject> potential = Maps.filterKeys(dataSet,
- createIsContainedPredicate(path));
- for (Entry<InstanceIdentifier<? extends DataObject>, DataObject> entry : potential.entrySet()) {
- try {
- builder.putAll(DataObjectReadingUtil.readData(entry.getValue(), (InstanceIdentifier) entry.getKey(),
- path));
- } catch (Exception e) {
- // FIXME : Log exception;
- }
- }
- return builder.build();
-
- }
-
- public class DelegatingDataReadRouter extends
- AbstractDataReadRouter<InstanceIdentifier<? extends DataObject>, DataObject> {
-
- private DataReader<InstanceIdentifier<? extends DataObject>, DataObject> delegate;
-
- @Override
- public DataObject readConfigurationData(final InstanceIdentifier<? extends DataObject> path) {
- return delegate.readConfigurationData(path);
- }
-
- public void setDelegate(final DataReader<InstanceIdentifier<? extends DataObject>, DataObject> delegate) {
- this.delegate = delegate;
- }
-
- @Override
- public DataObject readOperationalData(final InstanceIdentifier<? extends DataObject> path) {
- return delegate.readOperationalData(path);
- }
-
- @Override
- protected DataObject merge(final InstanceIdentifier<? extends DataObject> path, final Iterable<DataObject> data) {
- throw new UnsupportedOperationException("Not supported");
- }
-
- @Override
- public Registration registerConfigurationReader(
- final InstanceIdentifier<? extends DataObject> path,
- final DataReader<InstanceIdentifier<? extends DataObject>, DataObject> reader) {
- throw new UnsupportedOperationException("Not supported");
- }
-
- @Override
- public Registration registerOperationalReader(
- final InstanceIdentifier<? extends DataObject> path,
- final DataReader<InstanceIdentifier<? extends DataObject>, DataObject> reader) {
- throw new UnsupportedOperationException("Not supported");
- }
- }
-
- @Override
- protected ImmutableList<DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject>> affectedCommitHandlers(
- final Set<InstanceIdentifier<? extends DataObject>> paths) {
- ImmutableList.Builder<DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject>> handlersBuilder = ImmutableList.builder();
- return handlersBuilder //
- .add(rootCommitHandler) //
- .addAll(super.affectedCommitHandlers(paths)) //
- .build();
- }
-
- public void setRootCommitHandler(final DataCommitHandler<InstanceIdentifier<? extends DataObject>, DataObject> commitHandler) {
- rootCommitHandler = commitHandler;
- }
-
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.impl;
-
-import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
-import org.opendaylight.controller.md.sal.common.impl.service.AbstractDataTransaction;
-import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.ListenerRegistry;
-import org.opendaylight.yangtools.yang.binding.DataObject;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-
-@Deprecated
-public class DataTransactionImpl extends AbstractDataTransaction<InstanceIdentifier<? extends DataObject>, DataObject>
- implements DataModificationTransaction {
- private final ListenerRegistry<DataTransactionListener> listeners = new ListenerRegistry<DataTransactionListener>();
-
-
-
- public DataTransactionImpl(Object identifier,DataBrokerImpl dataBroker) {
- super(identifier,dataBroker);
- }
-
- @Override
- public ListenerRegistration<DataTransactionListener> registerListener(DataTransactionListener listener) {
- return listeners.register(listener);
- }
-
- @Override
- protected void onStatusChange(TransactionStatus status) {
- for (ListenerRegistration<DataTransactionListener> listenerRegistration : listeners) {
- listenerRegistration.getInstance().onStatusUpdated(this, status);
- }
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.impl;
-
-import static com.google.common.base.Preconditions.checkState;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
-import com.google.common.util.concurrent.UncheckedExecutionException;
-import java.util.EventListener;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
-import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
-import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
-import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
-import org.opendaylight.controller.sal.binding.codegen.RpcIsNotRoutedException;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper;
-import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
-import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.ListenerRegistry;
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RpcProviderRegistryImpl implements RpcProviderRegistry, RouteChangePublisher<RpcContextIdentifier, InstanceIdentifier<?>> {
-
- private RuntimeCodeGenerator rpcFactory = SingletonHolder.RPC_GENERATOR_IMPL;
-
- // cache of proxy objects where each value in the map corresponds to a specific RpcService
- private final LoadingCache<Class<? extends RpcService>, RpcService> publicProxies = CacheBuilder.newBuilder().weakKeys().
- build(new CacheLoader<Class<? extends RpcService>, RpcService>() {
- @Override
- public RpcService load(final Class<? extends RpcService> type) {
- final RpcService proxy = rpcFactory.getDirectProxyFor(type);
- LOG.debug("Created {} as public proxy for {} in {}", proxy, type.getSimpleName(), this);
- return proxy;
- }
- });
-
- private final Cache<Class<? extends RpcService>, RpcRouter<?>> rpcRouters = CacheBuilder.newBuilder().weakKeys()
- .build();
-
- private final ListenerRegistry<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> routeChangeListeners = ListenerRegistry
- .create();
- private final ListenerRegistry<RouterInstantiationListener> routerInstantiationListener = ListenerRegistry.create();
-
- private final static Logger LOG = LoggerFactory.getLogger(RpcProviderRegistryImpl.class);
-
- private final String name;
-
- private final ListenerRegistry<GlobalRpcRegistrationListener> globalRpcListeners = ListenerRegistry.create();
-
- public String getName() {
- return name;
- }
-
- public RpcProviderRegistryImpl(final String name) {
- super();
- this.name = name;
- }
-
- @Override
- public final <T extends RpcService> RoutedRpcRegistration<T> addRoutedRpcImplementation(final Class<T> type,
- final T implementation) throws IllegalStateException {
- return getRpcRouter(type).addRoutedRpcImplementation(implementation);
- }
-
- @Override
- public final <T extends RpcService> RpcRegistration<T> addRpcImplementation(final Class<T> type, final T implementation) {
-
- // FIXME: This should be well documented - addRpcImplementation for
- // routed RPCs
- try {
- // Note: If RPC is really global, expected count of registrations
- // of this method is really low.
- RpcRouter<T> potentialRouter = getRpcRouter(type);
- checkState(potentialRouter.getDefaultService() == null,
- "Default service for routed RPC already registered.");
- return potentialRouter.registerDefaultService(implementation);
- } catch (RpcIsNotRoutedException e) {
- // NOOP - we could safely continue, since RPC is not routed
- // so we fallback to global routing.
- LOG.debug("RPC is not routed. Using global registration.",e);
- }
- T publicProxy = getRpcService(type);
- RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
- checkState(currentDelegate == null, "Rpc service is already registered");
- LOG.debug("Registering {} as global implementation of {} in {}", implementation, type.getSimpleName(), this);
- RuntimeCodeHelper.setDelegate(publicProxy, implementation);
- notifyGlobalRpcAdded(type);
- return new RpcProxyRegistration<T>(type, implementation, this);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public final <T extends RpcService> T getRpcService(final Class<T> type) {
- return (T) publicProxies.getUnchecked(type);
- }
-
-
- public <T extends RpcService> RpcRouter<T> getRpcRouter(final Class<T> type) {
- try {
- final AtomicBoolean created = new AtomicBoolean(false);
- @SuppressWarnings( "unchecked")
- // LoadingCache is unsuitable for RpcRouter since we need to distinguish
- // first creation of RPC Router, so that is why
- // we are using normal cache with load API and shared AtomicBoolean
- // for this call, which will be set to true if router was created.
- RpcRouter<T> router = (RpcRouter<T>) rpcRouters.get(type,new Callable<RpcRouter<?>>() {
-
- @Override
- public org.opendaylight.controller.sal.binding.api.rpc.RpcRouter<?> call() {
- RpcRouter<?> router = rpcFactory.getRouterFor(type, name);
- router.registerRouteChangeListener(new RouteChangeForwarder<T>(type));
- LOG.debug("Registering router {} as global implementation of {} in {}", router, type.getSimpleName(), this);
- RuntimeCodeHelper.setDelegate(getRpcService(type), router.getInvocationProxy());
- created.set(true);
- return router;
- }
- });
- if(created.get()) {
- notifyListenersRoutedCreated(router);
- }
- return router;
- } catch (ExecutionException | UncheckedExecutionException e) {
- // We rethrow Runtime Exceptions which were wrapped by
- // Execution Exceptions
- // otherwise we throw IllegalStateException with original
- Throwables.propagateIfPossible(e.getCause());
- throw new IllegalStateException("Could not load RPC Router for "+type.getName(),e);
- }
- }
-
- private void notifyGlobalRpcAdded(final Class<? extends RpcService> type) {
- for(ListenerRegistration<GlobalRpcRegistrationListener> listener : globalRpcListeners) {
- try {
- listener.getInstance().onGlobalRpcRegistered(type);
- } catch (Exception e) {
- LOG.error("Unhandled exception during invoking listener {}", e);
- }
- }
-
- }
-
- private void notifyListenersRoutedCreated(final RpcRouter<?> router) {
-
- for (ListenerRegistration<RouterInstantiationListener> listener : routerInstantiationListener) {
- try {
- listener.getInstance().onRpcRouterCreated(router);
- } catch (Exception e) {
- LOG.error("Unhandled exception during invoking listener {}", e);
- }
- }
-
- }
-
- public ListenerRegistration<RouterInstantiationListener> registerRouterInstantiationListener(
- final RouterInstantiationListener listener) {
- ListenerRegistration<RouterInstantiationListener> reg = routerInstantiationListener.register(listener);
- try {
- for (RpcRouter<?> router : rpcRouters.asMap().values()) {
- listener.onRpcRouterCreated(router);
- }
- } catch (Exception e) {
- LOG.error("Unhandled exception during invoking listener {}", e);
- }
- return reg;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public <L extends RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> ListenerRegistration<L> registerRouteChangeListener(
- final L listener) {
- return (ListenerRegistration<L>) routeChangeListeners.register(listener);
- }
-
- public RuntimeCodeGenerator getRpcFactory() {
- return rpcFactory;
- }
-
- public void setRpcFactory(final RuntimeCodeGenerator rpcFactory) {
- this.rpcFactory = rpcFactory;
- }
-
- public interface RouterInstantiationListener extends EventListener {
- void onRpcRouterCreated(RpcRouter<?> router);
- }
-
- public ListenerRegistration<GlobalRpcRegistrationListener> registerGlobalRpcRegistrationListener(final GlobalRpcRegistrationListener listener) {
- return globalRpcListeners.register(listener);
- }
-
- public interface GlobalRpcRegistrationListener extends EventListener {
- void onGlobalRpcRegistered(Class<? extends RpcService> cls);
- void onGlobalRpcUnregistered(Class<? extends RpcService> cls);
-
- }
-
- private final class RouteChangeForwarder<T extends RpcService> implements RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>> {
- private final Class<T> type;
-
- RouteChangeForwarder(final Class<T> type) {
- this.type = type;
- }
-
- @Override
- public void onRouteChange(final RouteChange<Class<? extends BaseIdentity>, InstanceIdentifier<?>> change) {
- Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> announcements = new HashMap<>();
- for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getAnnouncements()
- .entrySet()) {
- RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
- announcements.put(key, entry.getValue());
- }
- Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> removals = new HashMap<>();
- for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getRemovals()
- .entrySet()) {
- RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
- removals.put(key, entry.getValue());
- }
- RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> toPublish = RoutingUtils
- .<RpcContextIdentifier, InstanceIdentifier<?>> change(announcements, removals);
- for (ListenerRegistration<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> listener : routeChangeListeners) {
- try {
- listener.getInstance().onRouteChange(toPublish);
- } catch (Exception e) {
- LOG.error("Unhandled exception during invoking listener",listener.getInstance(),e);
- }
- }
- }
- }
-
- private static final class RpcProxyRegistration<T extends RpcService> extends AbstractObjectRegistration<T> implements RpcRegistration<T> {
- private final RpcProviderRegistryImpl registry;
- private final Class<T> serviceType;
-
- RpcProxyRegistration(final Class<T> type, final T service, final RpcProviderRegistryImpl registry) {
- super(service);
- this.registry = Preconditions.checkNotNull(registry);
- this.serviceType = type;
- }
-
- @Override
- public Class<T> getServiceType() {
- return serviceType;
- }
-
- @Override
- protected void removeRegistration() {
- T publicProxy = registry.getRpcService(serviceType);
- RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
- if (currentDelegate == getInstance()) {
- RuntimeCodeHelper.setDelegate(publicProxy, null);
- }
- }
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-
-package org.opendaylight.controller.sal.binding.impl;
-
-import org.opendaylight.yangtools.yang.binding.RpcService;
-import org.osgi.framework.ServiceRegistration;
-
-@SuppressWarnings("all")
-public class RpcProxyContext {
- public RpcProxyContext(final Class<? extends RpcService> proxyClass) {
- this.proxyClass = proxyClass;
- }
-
- protected final Class<? extends RpcService> proxyClass;
-
- protected RpcService _proxy;
-
- public RpcService getProxy() {
- return this._proxy;
- }
-
- public void setProxy(final RpcService proxy) {
- this._proxy = proxy;
- }
-
- protected ServiceRegistration<? extends RpcService> _registration;
-
- public ServiceRegistration<? extends RpcService> getRegistration() {
- return this._registration;
- }
-
- public void setRegistration(final ServiceRegistration<? extends RpcService> registration) {
- this._registration = registration;
- }
-}
\ No newline at end of file
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.impl.util;
-
-import org.opendaylight.controller.md.sal.common.impl.routing.AbstractDataReadRouter;
-import org.opendaylight.yangtools.yang.binding.DataObject;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-
-public class BindingAwareDataReaderRouter extends AbstractDataReadRouter<InstanceIdentifier<? extends DataObject>,DataObject> {
- @Override
- protected DataObject merge(final InstanceIdentifier<? extends DataObject> path, final Iterable<DataObject> data) {
- return data.iterator().next();
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.spi;
-
-import java.util.Set;
-
-import org.opendaylight.controller.sal.binding.api.NotificationListener;
-import org.opendaylight.yangtools.yang.binding.Notification;
-
-public interface NotificationInvokerFactory {
-
- NotificationInvoker invokerFor(org.opendaylight.yangtools.yang.binding.NotificationListener instance);
-
- public interface NotificationInvoker {
-
- Set<Class<? extends Notification>> getSupportedNotifications();
-
- NotificationListener<Notification> getInvocationProxy();
-
- public abstract void close();
-
- org.opendaylight.yangtools.yang.binding.NotificationListener getDelegate();
-
- }
-}
}
}
+ augment "/config:modules/config:module/config:configuration" {
+ case binding-notification-broker {
+ when "/config:modules/config:module/config:type = 'binding-notification-broker'";
+ container notification-adapter {
+ uses config:service-ref {
+ refine type {
+ mandatory false;
+ config:required-identity binding-new-notification-service;
+ }
+ }
+ }
+
+ container notification-publish-adapter {
+ uses config:service-ref {
+ refine type {
+ mandatory false;
+ config:required-identity binding-new-notification-publish-service;
+ }
+ }
+ }
+ }
+ }
+
augment "/config:modules/config:module/config:state" {
case binding-notification-broker {
when "/config:modules/config:module/config:type = 'binding-notification-broker'";
+++ /dev/null
-package org.opendaylight.controller.md.sal.binding.impl.test;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
-import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.TOP_BAR_KEY;
-import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.TOP_FOO_KEY;
-import static org.opendaylight.controller.md.sal.test.model.util.ListsBindingUtils.path;
-
-import com.google.common.base.Throwables;
-import java.util.Arrays;
-import javassist.ClassPool;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.mockito.Mockito;
-import org.opendaylight.controller.md.sal.binding.compat.HeliumRpcProviderRegistry;
-import org.opendaylight.controller.md.sal.binding.impl.BindingDOMRpcProviderServiceAdapter;
-import org.opendaylight.controller.md.sal.binding.impl.BindingDOMRpcServiceAdapter;
-import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodec;
-import org.opendaylight.controller.md.sal.binding.test.AbstractSchemaAwareTest;
-import org.opendaylight.controller.md.sal.dom.broker.impl.DOMRpcRouter;
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
-import org.opendaylight.controller.sal.binding.api.RpcConsumerRegistry;
-import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
-import org.opendaylight.controller.sal.binding.codegen.RpcIsNotRoutedException;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.bi.ba.rpcservice.rev140701.OpendaylightTestRpcServiceService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.list.rev140701.two.level.list.TopLevelList;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.rpc.routing.rev140701.OpendaylightTestRoutedRpcService;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.test.rpc.routing.rev140701.TestContext;
-import org.opendaylight.yangtools.binding.data.codec.gen.impl.DataObjectSerializerGenerator;
-import org.opendaylight.yangtools.binding.data.codec.gen.impl.StreamWriterGenerator;
-import org.opendaylight.yangtools.binding.data.codec.impl.BindingNormalizedNodeCodecRegistry;
-import org.opendaylight.yangtools.sal.binding.generator.impl.GeneratedClassLoadingStrategy;
-import org.opendaylight.yangtools.sal.binding.generator.util.JavassistUtils;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.YangModuleInfo;
-import org.opendaylight.yangtools.yang.binding.util.BindingReflections;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-
-
-public class RpcProviderRegistryTest extends AbstractSchemaAwareTest {
-
- private static InstanceIdentifier<TopLevelList> FOO_PATH = path(TOP_FOO_KEY);
- private static InstanceIdentifier<TopLevelList> BAR_PATH = path(TOP_BAR_KEY);
- private static RpcContextIdentifier ROUTING_CONTEXT = RpcContextIdentifier.contextFor(OpendaylightTestRoutedRpcService.class, TestContext.class);
-
- private RpcProviderRegistry rpcRegistry;
-
-
- @Override
- protected Iterable<YangModuleInfo> getModuleInfos() {
- try {
- return Arrays.asList(
- BindingReflections.getModuleInfo(TopLevelList.class),
- BindingReflections.getModuleInfo(OpendaylightTestRoutedRpcService.class),
- BindingReflections.getModuleInfo(OpendaylightTestRpcServiceService.class));
- } catch (final Exception e) {
- throw Throwables.propagate(e);
- }
- }
-
- @Override
- protected void setupWithSchema(final SchemaContext context) {
- final DataObjectSerializerGenerator generator = StreamWriterGenerator.create(JavassistUtils.forClassPool(ClassPool.getDefault()));
- final BindingNormalizedNodeCodecRegistry codecRegistry = new BindingNormalizedNodeCodecRegistry(generator);
- final GeneratedClassLoadingStrategy classLoadingStrategy = GeneratedClassLoadingStrategy.getTCCLClassLoadingStrategy();
- final BindingToNormalizedNodeCodec codec = new BindingToNormalizedNodeCodec(classLoadingStrategy, codecRegistry);
- final DOMRpcRouter domRpcRegistry = new DOMRpcRouter();
- domRpcRegistry.onGlobalContextUpdated(context);
- codec.onGlobalContextUpdated(context);
- final RpcConsumerRegistry consumer = new BindingDOMRpcServiceAdapter(domRpcRegistry, codec);
- final BindingDOMRpcProviderServiceAdapter provider = new BindingDOMRpcProviderServiceAdapter( domRpcRegistry,codec);
- rpcRegistry = new HeliumRpcProviderRegistry(consumer,provider);
- }
-
- @Test
- public void testGlobalRpcRegistrations() throws Exception {
- final OpendaylightTestRpcServiceService one = Mockito.mock(OpendaylightTestRpcServiceService.class);
- final OpendaylightTestRpcServiceService two = Mockito.mock(OpendaylightTestRpcServiceService.class);
-
- final RpcRegistration<OpendaylightTestRpcServiceService> regOne = rpcRegistry.addRpcImplementation(OpendaylightTestRpcServiceService.class, one);
- assertNotNull(regOne);
- rpcRegistry.addRpcImplementation(OpendaylightTestRpcServiceService.class, two);
- regOne.close();
- final RpcRegistration<OpendaylightTestRpcServiceService> regTwo = rpcRegistry.addRpcImplementation(OpendaylightTestRpcServiceService.class, two);
- assertNotNull(regTwo);
- }
-
-
- @Test
- @Ignore
- public void nonRoutedRegisteredAsRouted() {
- final OpendaylightTestRpcServiceService one = Mockito.mock(OpendaylightTestRpcServiceService.class);
- try {
- final RoutedRpcRegistration<OpendaylightTestRpcServiceService> reg = rpcRegistry.addRoutedRpcImplementation(OpendaylightTestRpcServiceService.class, one);
- reg.registerPath(null, BAR_PATH);
- fail("RpcIsNotRoutedException should be thrown");
- } catch (final RpcIsNotRoutedException e) {
- assertNotNull(e.getMessage());
- } catch (final Exception e) {
- fail("RpcIsNotRoutedException should be thrown");
- }
-
- }
-
-}
import org.opendaylight.controller.md.sal.dom.broker.impl.DOMNotificationRouter;
import org.opendaylight.controller.md.sal.dom.broker.impl.SerializedDOMDataBroker;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
-import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
import org.opendaylight.controller.sal.binding.test.util.MockSchemaService;
import org.opendaylight.controller.sal.core.api.model.SchemaService;
import org.opendaylight.controller.sal.core.spi.data.DOMStore;
}
public NotificationService createNotificationService() {
- return new BindingDOMNotificationServiceAdapter(bindingToNormalized.getCodecRegistry(), domNotificationRouter,
- SingletonHolder.INVOKER_FACTORY);
+ return new BindingDOMNotificationServiceAdapter(bindingToNormalized.getCodecRegistry(), domNotificationRouter);
}
public NotificationPublishService createNotificationPublishService() {
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import java.util.ArrayList;
-import java.util.List;
-import javassist.ClassPool;
-import org.junit.Before;
-import org.junit.Test;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
-import org.opendaylight.controller.sal.binding.api.rpc.RpcRoutingTable;
-import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory;
-import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker;
-import org.opendaylight.controller.sal.binding.test.mock.BarListener;
-import org.opendaylight.controller.sal.binding.test.mock.BarUpdate;
-import org.opendaylight.controller.sal.binding.test.mock.FlowDelete;
-import org.opendaylight.controller.sal.binding.test.mock.FooListener;
-import org.opendaylight.controller.sal.binding.test.mock.FooService;
-import org.opendaylight.controller.sal.binding.test.mock.FooUpdate;
-import org.opendaylight.controller.sal.binding.test.mock.ReferencableObject;
-import org.opendaylight.controller.sal.binding.test.mock.ReferencableObjectKey;
-import org.opendaylight.controller.sal.binding.test.mock.SimpleInput;
-import org.opendaylight.yangtools.yang.binding.Augmentation;
-import org.opendaylight.yangtools.yang.binding.BaseIdentity;
-import org.opendaylight.yangtools.yang.binding.DataContainer;
-import org.opendaylight.yangtools.yang.binding.DataObject;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-
-public class DefaultRuntimeCodeGeneratorTest {
-
- private RuntimeCodeGenerator codeGenerator;
- private NotificationInvokerFactory invokerFactory;
-
- @Before
- public void initialize() {
- this.codeGenerator = new DefaultRuntimeCodeGenerator(ClassPool.getDefault());
- this.invokerFactory = codeGenerator.getInvokerFactory();
- }
-
- @Test
- public void testGenerateDirectProxy() {
- FooService product = codeGenerator.getDirectProxyFor(FooService.class);
- assertNotNull(product);
- }
-
- @Test
- public void testGenerateRouter() throws Exception {
- RpcRouter<FooService> product = codeGenerator.getRouterFor(FooService.class,"test");
- assertNotNull(product);
- assertNotNull(product.getInvocationProxy());
-
- assertEquals("2 fields should be generated.", 2, product.getInvocationProxy().getClass().getFields().length);
-
- verifyRouting(product);
- }
-
- @Test
- public void testInvoker() throws Exception {
-
- FooListenerImpl fooListener = new FooListenerImpl();
-
- NotificationInvoker invokerFoo = invokerFactory.invokerFor(fooListener);
-
-
- assertSame(fooListener,invokerFoo.getDelegate());
- assertNotNull(invokerFoo.getSupportedNotifications());
- assertEquals(1, invokerFoo.getSupportedNotifications().size());
- assertNotNull(invokerFoo.getInvocationProxy());
-
- FooUpdateImpl fooOne = new FooUpdateImpl();
- invokerFoo.getInvocationProxy().onNotification(fooOne);
-
- assertEquals(1, fooListener.receivedFoos.size());
- assertSame(fooOne, fooListener.receivedFoos.get(0));
-
- CompositeListenerImpl composite = new CompositeListenerImpl();
-
- NotificationInvoker invokerComposite = invokerFactory.invokerFor(composite);
-
- assertNotNull(invokerComposite.getSupportedNotifications());
- assertEquals(3, invokerComposite.getSupportedNotifications().size());
- assertNotNull(invokerComposite.getInvocationProxy());
-
- invokerComposite.getInvocationProxy().onNotification(fooOne);
-
- assertEquals(1, composite.receivedFoos.size());
- assertSame(fooOne, composite.receivedFoos.get(0));
-
- assertEquals(0, composite.receivedBars.size());
-
- BarUpdateImpl barOne = new BarUpdateImpl();
-
- invokerComposite.getInvocationProxy().onNotification(barOne);
-
- assertEquals(1, composite.receivedFoos.size());
- assertEquals(1, composite.receivedBars.size());
- assertSame(barOne, composite.receivedBars.get(0));
-
- }
-
- private void verifyRouting(final RpcRouter<FooService> product) {
- assertNotNull("Routing table should be initialized", product.getRoutingTable(BaseIdentity.class));
-
- RpcRoutingTable<BaseIdentity, FooService> routingTable = product.getRoutingTable(BaseIdentity.class);
-
- int servicesCount = 2;
- int instancesPerService = 3;
-
- InstanceIdentifier<?>[][] identifiers = identifiers(servicesCount, instancesPerService);
- FooService service[] = new FooService[] { mock(FooService.class, "Instance 0"),
- mock(FooService.class, "Instance 1") };
-
- for (int i = 0; i < service.length; i++) {
- for (InstanceIdentifier<?> instance : identifiers[i]) {
- routingTable.updateRoute(instance, service[i]);
- }
- }
-
- assertEquals("All instances should be registered.", servicesCount * instancesPerService, routingTable
- .getRoutes().size());
-
- SimpleInput[] instance_0_input = new SimpleInputImpl[] { new SimpleInputImpl(identifiers[0][0]),
- new SimpleInputImpl(identifiers[0][1]), new SimpleInputImpl(identifiers[0][2]) };
-
- SimpleInput[] instance_1_input = new SimpleInputImpl[] { new SimpleInputImpl(identifiers[1][0]),
- new SimpleInputImpl(identifiers[1][1]), new SimpleInputImpl(identifiers[1][2]) };
-
- // We test sending mock messages
-
- product.getInvocationProxy().simple(instance_0_input[0]);
- verify(service[0]).simple(instance_0_input[0]);
-
- product.getInvocationProxy().simple(instance_0_input[1]);
- product.getInvocationProxy().simple(instance_0_input[2]);
-
- verify(service[0]).simple(instance_0_input[1]);
- verify(service[0]).simple(instance_0_input[2]);
-
- product.getInvocationProxy().simple(instance_1_input[0]);
-
- // We should have call to instance 1
- verify(service[1]).simple(instance_1_input[0]);
-
- /*
- * Generated RPC service should throw illegalArgumentException
- * with message if rpc input is null.
- */
- try {
- product.getInvocationProxy().simple(null);
- fail("Generated RPC router should throw IllegalArgumentException on null input");
- } catch (IllegalArgumentException e){
- assertNotNull(e.getMessage());
- }
-
-
- /*
- * Generated RPC service should throw illegalArgumentException
- * with message if rpc route is null.
- */
- try {
- SimpleInput withoutValue = new SimpleInputImpl(null);
- product.getInvocationProxy().simple(withoutValue);
- fail("Generated RPC router should throw IllegalArgumentException on null value for route");
- } catch (IllegalArgumentException e){
- assertNotNull(e.getMessage());
- }
-
- }
-
- private InstanceIdentifier<?>[][] identifiers(final int serviceSize, final int instancesPerService) {
- InstanceIdentifier<?>[][] ret = new InstanceIdentifier[serviceSize][];
- int service = 0;
- for (int i = 0; i < serviceSize; i++) {
-
- InstanceIdentifier<?>[] instanceIdentifiers = new InstanceIdentifier[instancesPerService];
- ret[i] = instanceIdentifiers;
- for (int id = 0; id < instancesPerService; id++) {
- instanceIdentifiers[id] = referencableIdentifier(service * instancesPerService + id);
- }
- service++;
- }
-
- return ret;
- }
-
- private InstanceIdentifier<?> referencableIdentifier(final int i) {
- return InstanceIdentifier.builder(ReferencableObject.class, new ReferencableObjectKey(i)).build();
- }
-
- private static class SimpleInputImpl implements SimpleInput {
- private final InstanceIdentifier<?> identifier;
-
- public SimpleInputImpl(final InstanceIdentifier<?> _identifier) {
- this.identifier = _identifier;
- }
-
- @Override
- public <E extends Augmentation<SimpleInput>> E getAugmentation(final Class<E> augmentationType) {
- return null;
- }
-
- @Override
- public InstanceIdentifier<?> getIdentifier() {
- return this.identifier;
- }
-
- @Override
- public Class<? extends DataObject> getImplementedInterface() {
- return SimpleInput.class;
- }
- }
-
- private static class FooUpdateImpl implements FooUpdate {
- @Override
- public Class<? extends DataContainer> getImplementedInterface() {
- return FooUpdate.class;
- }
- }
-
- private static class BarUpdateImpl implements BarUpdate {
- @Override
- public Class<? extends DataContainer> getImplementedInterface() {
- return BarUpdate.class;
- }
-
- @Override
- public InstanceIdentifier<?> getInheritedIdentifier() {
- return null;
- }
- }
-
- private static class FooListenerImpl implements FooListener {
-
- List<FooUpdate> receivedFoos = new ArrayList<>();
-
- @Override
- public void onFooUpdate(final FooUpdate notification) {
- receivedFoos.add(notification);
- }
-
- }
-
- private static class CompositeListenerImpl extends FooListenerImpl implements BarListener {
-
- List<BarUpdate> receivedBars = new ArrayList<>();
- List<FlowDelete> receivedDeletes = new ArrayList<>();
-
- @Override
- public void onBarUpdate(final BarUpdate notification) {
- receivedBars.add(notification);
- }
-
- @Override
- public void onFlowDelete(final FlowDelete notification) {
- receivedDeletes.add(notification);
- }
-
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-
-package org.opendaylight.controller.sal.binding.codegen.impl;
-
-import com.google.common.util.concurrent.ListeningExecutorService;
-import java.lang.reflect.Field;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Ignore
-public class SingletonHolderTest {
- private static final Logger logger = LoggerFactory.getLogger(SingletonHolderTest.class);
-
- @Test
- public void testNotificationExecutor() throws Exception {
- ListeningExecutorService executor = SingletonHolder.getDefaultNotificationExecutor();
- ThreadPoolExecutor tpExecutor = (ThreadPoolExecutor) setAccessible(executor.getClass().getDeclaredField("delegate")).get(executor);
- BlockingQueue<Runnable> queue = tpExecutor.getQueue();
-
- for (int idx = 0; idx < 100; idx++) {
- final int idx2 = idx;
- logger.info("Adding {}\t{}\t{}", idx, queue.size(), tpExecutor.getActiveCount());
- executor.execute(new Runnable() {
-
- @Override
- public void run() {
- logger.info("in {}", idx2);
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- logger.info("out {}", idx2);
- }
- });
- }
- executor.shutdown();
- executor.awaitTermination(10, TimeUnit.SECONDS);
- }
-
- private static Field setAccessible(Field field) {
- field.setAccessible(true);
- return field;
- }
-}
import javassist.ClassPool;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.md.sal.binding.api.MountPointService;
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
+import org.opendaylight.controller.md.sal.binding.api.NotificationService;
+import org.opendaylight.controller.md.sal.binding.compat.HeliumNotificationProviderServiceAdapter;
import org.opendaylight.controller.md.sal.binding.compat.HeliumRpcProviderRegistry;
import org.opendaylight.controller.md.sal.binding.compat.HydrogenDataBrokerAdapter;
import org.opendaylight.controller.md.sal.binding.compat.HydrogenMountProvisionServiceAdapter;
import org.opendaylight.controller.md.sal.binding.impl.BindingDOMDataBrokerAdapter;
import org.opendaylight.controller.md.sal.binding.impl.BindingDOMMountPointServiceAdapter;
+import org.opendaylight.controller.md.sal.binding.impl.BindingDOMNotificationPublishServiceAdapter;
+import org.opendaylight.controller.md.sal.binding.impl.BindingDOMNotificationServiceAdapter;
import org.opendaylight.controller.md.sal.binding.impl.BindingDOMRpcProviderServiceAdapter;
import org.opendaylight.controller.md.sal.binding.impl.BindingDOMRpcServiceAdapter;
import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodec;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
import org.opendaylight.controller.md.sal.dom.api.DOMRpcProviderService;
import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
+import org.opendaylight.controller.md.sal.dom.broker.impl.DOMNotificationRouter;
import org.opendaylight.controller.md.sal.dom.broker.impl.DOMRpcRouter;
import org.opendaylight.controller.md.sal.dom.broker.impl.SerializedDOMDataBroker;
import org.opendaylight.controller.md.sal.dom.broker.impl.mount.DOMMountPointServiceImpl;
import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
import org.opendaylight.controller.sal.binding.api.mount.MountProviderService;
-import org.opendaylight.controller.sal.binding.impl.NotificationBrokerImpl;
import org.opendaylight.controller.sal.binding.impl.RootBindingAwareBroker;
import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;
import org.opendaylight.controller.sal.core.api.BrokerService;
private RootBindingAwareBroker baBrokerImpl;
- private NotificationBrokerImpl baNotifyImpl;
+ private HeliumNotificationProviderServiceAdapter baNotifyImpl;
private BrokerImpl biBrokerImpl;
private BindingDOMRpcProviderServiceAdapter baProviderRpc;
private DOMRpcRouter domRouter;
+ private NotificationPublishService publishService;
+
+ private NotificationService listenService;
+
+ private DOMNotificationPublishService domPublishService;
+
+ private DOMNotificationService domListenService;
+
public DOMDataBroker getDomAsyncDataBroker() {
public void startBindingNotificationBroker() {
checkState(executor != null);
- baNotifyImpl = new NotificationBrokerImpl(executor);
+ final DOMNotificationRouter router = DOMNotificationRouter.create(16);
+ domPublishService = router;
+ domListenService = router;
+ publishService = new BindingDOMNotificationPublishServiceAdapter(codec, domPublishService);
+ listenService = new BindingDOMNotificationServiceAdapter(codec, domListenService);
+ baNotifyImpl = new HeliumNotificationProviderServiceAdapter(publishService,listenService);
}
<module>
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding:impl">prefix:binding-notification-broker</type>
<name>binding-notification-broker</name>
+ <notification-adapter xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding:impl">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding:impl">prefix:binding-new-notification-service</type>
+ <name>binding-notification-adapter</name>
+ </notification-adapter>
+ <notification-publish-adapter xmlns="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding:impl">
+ <type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding:impl">prefix:binding-new-notification-publish-service</type>
+ <name>binding-notification-publish-adapter</name>
+ </notification-publish-adapter>
</module>
<module>
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:binding:impl">prefix:binding-broker-impl</type>
<module>
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom:impl">prefix:dom-inmemory-data-broker</type>
<name>inmemory-data-broker</name>
+
<schema-service>
<type xmlns:dom="urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom">dom:schema-service</type>
<name>yang-schema-service</name>
</schema-service>
+
+ <config-data-store>
+ <type xmlns:config-dom-store-spi="urn:opendaylight:params:xml:ns:yang:controller:md:sal:core:spi:config-dom-store">config-dom-store-spi:config-dom-datastore</type>
+ <name>config-store-service</name>
+ </config-data-store>
+
+ <operational-data-store>
+ <type xmlns:operational-dom-store-spi="urn:opendaylight:params:xml:ns:yang:controller:md:sal:core:spi:operational-dom-store">operational-dom-store-spi:operational-dom-datastore</type>
+ <name>operational-store-service</name>
+ </operational-data-store>
</module>
<module>
<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom:impl">prefix:dom-broker-impl</type>
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import static com.google.common.base.Preconditions.checkState;
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBrokerExtension;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeService;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeIdentifier;
+import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
+import org.opendaylight.controller.sal.core.spi.data.DOMStore;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTreeChangePublisher;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractDOMBroker extends AbstractDOMTransactionFactory<DOMStore>
+ implements DOMDataBroker, AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractDOMBroker.class);
+
+ private final AtomicLong txNum = new AtomicLong();
+ private final AtomicLong chainNum = new AtomicLong();
+ private final Map<Class<? extends DOMDataBrokerExtension>, DOMDataBrokerExtension> extensions;
+ private volatile AutoCloseable closeable;
+
+ protected AbstractDOMBroker(final Map<LogicalDatastoreType, DOMStore> datastores) {
+ super(datastores);
+
+ boolean treeChange = true;
+ for (DOMStore ds : datastores.values()) {
+ if (!(ds instanceof DOMStoreTreeChangePublisher)) {
+ treeChange = false;
+ break;
+ }
+ }
+
+ if (treeChange) {
+ extensions = ImmutableMap.<Class<? extends DOMDataBrokerExtension>, DOMDataBrokerExtension>of(DOMDataTreeChangeService.class, new DOMDataTreeChangeService() {
+ @Override
+ public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerDataTreeChangeListener(final DOMDataTreeIdentifier treeId, final L listener) {
+ DOMStore publisher = getTxFactories().get(treeId.getDatastoreType());
+ checkState(publisher != null, "Requested logical data store is not available.");
+
+ return ((DOMStoreTreeChangePublisher) publisher).registerTreeChangeListener(treeId.getRootIdentifier(), listener);
+ }
+ });
+ } else {
+ extensions = Collections.emptyMap();
+ }
+ }
+
+ public void setCloseable(final AutoCloseable closeable) {
+ this.closeable = closeable;
+ }
+
+ @Override
+ public void close() {
+ super.close();
+
+ if (closeable != null) {
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ LOG.debug("Error closing instance", e);
+ }
+ }
+ }
+
+ @Override
+ protected Object newTransactionIdentifier() {
+ return "DOM-" + txNum.getAndIncrement();
+ }
+
+ @Override
+ public ListenerRegistration<DOMDataChangeListener> registerDataChangeListener(final LogicalDatastoreType store,
+ final YangInstanceIdentifier path, final DOMDataChangeListener listener, final DataChangeScope triggeringScope) {
+
+ DOMStore potentialStore = getTxFactories().get(store);
+ checkState(potentialStore != null, "Requested logical data store is not available.");
+ return potentialStore.registerChangeListener(path, listener, triggeringScope);
+ }
+
+ @Override
+ public Map<Class<? extends DOMDataBrokerExtension>, DOMDataBrokerExtension> getSupportedExtensions() {
+ return extensions;
+ }
+
+ @Override
+ public DOMTransactionChain createTransactionChain(final TransactionChainListener listener) {
+ checkNotClosed();
+
+ final Map<LogicalDatastoreType, DOMStoreTransactionChain> backingChains = new EnumMap<>(LogicalDatastoreType.class);
+ for (Map.Entry<LogicalDatastoreType, DOMStore> entry : getTxFactories().entrySet()) {
+ backingChains.put(entry.getKey(), entry.getValue().createTransactionChain());
+ }
+
+ final long chainId = chainNum.getAndIncrement();
+ LOG.debug("Transaction chain {} created with listener {}, backing store chains {}", chainId, listener,
+ backingChains);
+ return new DOMBrokerTransactionChain(chainId, backingChains, this, listener);
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import com.google.common.base.Preconditions;
+import java.util.Collection;
+import java.util.EnumMap;
+import java.util.Map;
+import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+public abstract class AbstractDOMBrokerTransaction<K, T extends DOMStoreTransaction> implements
+ AsyncTransaction<YangInstanceIdentifier, NormalizedNode<?, ?>> {
+
+ private Map<K, T> backingTxs;
+ private final Object identifier;
+ private final Map<LogicalDatastoreType, ? extends DOMStoreTransactionFactory> storeTxFactories;
+
+ /**
+ *
+ * Creates new composite Transactions.
+ *
+ * @param identifier
+ * Identifier of transaction.
+ */
+ protected AbstractDOMBrokerTransaction(final Object identifier, Map<LogicalDatastoreType, ? extends DOMStoreTransactionFactory> storeTxFactories) {
+ this.identifier = Preconditions.checkNotNull(identifier, "Identifier should not be null");
+ this.storeTxFactories = Preconditions.checkNotNull(storeTxFactories, "Store Transaction Factories should not be null");
+ this.backingTxs = new EnumMap(LogicalDatastoreType.class);
+ }
+
+ /**
+ * Returns subtransaction associated with supplied key.
+ *
+ * @param key
+ * @return
+ * @throws NullPointerException
+ * if key is null
+ * @throws IllegalArgumentException
+ * if no subtransaction is associated with key.
+ */
+ protected final T getSubtransaction(final K key) {
+ Preconditions.checkNotNull(key, "key must not be null.");
+
+ T ret = backingTxs.get(key);
+ if(ret == null){
+ ret = createTransaction(key);
+ backingTxs.put(key, ret);
+ }
+ Preconditions.checkArgument(ret != null, "No subtransaction associated with %s", key);
+ return ret;
+ }
+
+ protected abstract T createTransaction(final K key);
+
+ /**
+ * Returns immutable Iterable of all subtransactions.
+ *
+ */
+ protected Collection<T> getSubtransactions() {
+ return backingTxs.values();
+ }
+
+ @Override
+ public Object getIdentifier() {
+ return identifier;
+ }
+
+ protected void closeSubtransactions() {
+ /*
+ * We share one exception for all failures, which are added
+ * as supressedExceptions to it.
+ */
+ IllegalStateException failure = null;
+ for (T subtransaction : backingTxs.values()) {
+ try {
+ subtransaction.close();
+ } catch (Exception e) {
+ // If we did not allocated failure we allocate it
+ if (failure == null) {
+ failure = new IllegalStateException("Uncaught exception occured during closing transaction", e);
+ } else {
+ // We update it with additional exceptions, which occurred during error.
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ // If we have failure, we throw it at after all attempts to close.
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ protected DOMStoreTransactionFactory getTxFactory(K type){
+ return storeTxFactories.get(type);
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.CheckedFuture;
+import java.util.Collection;
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
+
+public abstract class AbstractDOMTransactionFactory<T extends DOMStoreTransactionFactory> implements AutoCloseable {
+ private static final AtomicIntegerFieldUpdater<AbstractDOMTransactionFactory> UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(AbstractDOMTransactionFactory.class, "closed");
+ private final Map<LogicalDatastoreType, T> storeTxFactories;
+ private volatile int closed = 0;
+
+ protected AbstractDOMTransactionFactory(final Map<LogicalDatastoreType, T> txFactories) {
+ this.storeTxFactories = new EnumMap<>(txFactories);
+ }
+
+ /**
+ * Implementations must return unique identifier for each and every call of
+ * this method;
+ *
+ * @return new Unique transaction identifier.
+ */
+ protected abstract Object newTransactionIdentifier();
+
+ /**
+ *
+ * @param transaction
+ * @param cohorts
+ * @return
+ */
+ protected abstract CheckedFuture<Void,TransactionCommitFailedException> submit(final DOMDataWriteTransaction transaction,
+ final Collection<DOMStoreThreePhaseCommitCohort> cohorts);
+
+ /**
+ *
+ * @return
+ */
+ public final DOMDataReadOnlyTransaction newReadOnlyTransaction() {
+ checkNotClosed();
+
+ return new DOMBrokerReadOnlyTransaction(newTransactionIdentifier(), storeTxFactories);
+ }
+
+
+ /**
+ *
+ * @return
+ */
+ public final DOMDataWriteTransaction newWriteOnlyTransaction() {
+ checkNotClosed();
+
+ return new DOMBrokerWriteOnlyTransaction(newTransactionIdentifier(), storeTxFactories, this);
+ }
+
+
+ /**
+ *
+ * @return
+ */
+ public final DOMDataReadWriteTransaction newReadWriteTransaction() {
+ checkNotClosed();
+
+ return new DOMBrokerReadWriteTransaction<>(newTransactionIdentifier(), storeTxFactories, this);
+ }
+
+ /**
+ * Convenience accessor of backing factories intended to be used only by
+ * finalization of this class.
+ *
+ * <b>Note:</b>
+ * Finalization of this class may want to access other functionality of
+ * supplied Transaction factories.
+ *
+ * @return Map of backing transaction factories.
+ */
+ protected final Map<LogicalDatastoreType, T> getTxFactories() {
+ return storeTxFactories;
+ }
+
+ /**
+ * Checks if instance is not closed.
+ *
+ * @throws IllegalStateException If instance of this class was closed.
+ *
+ */
+ protected final void checkNotClosed() {
+ Preconditions.checkState(closed == 0, "Transaction factory was closed. No further operations allowed.");
+ }
+
+ @Override
+ public void close() {
+ final boolean success = UPDATER.compareAndSet(this, 0, 1);
+ Preconditions.checkState(success, "Transaction factory was already closed");
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.CheckedFuture;
+import java.util.Map;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+public class DOMBrokerReadOnlyTransaction<T extends DOMStoreReadTransaction>
+ extends AbstractDOMBrokerTransaction<LogicalDatastoreType, T>
+ implements DOMDataReadOnlyTransaction {
+ /**
+ * Creates new composite Transactions.
+ *
+ * @param identifier Identifier of transaction.
+ */
+ protected DOMBrokerReadOnlyTransaction(Object identifier, Map storeTxFactories) {
+ super(identifier, storeTxFactories);
+ }
+
+ @Override
+ public CheckedFuture<Optional<NormalizedNode<?,?>>, ReadFailedException> read(
+ final LogicalDatastoreType store, final YangInstanceIdentifier path) {
+ return getSubtransaction(store).read(path);
+ }
+
+ @Override
+ public CheckedFuture<Boolean, ReadFailedException> exists(
+ final LogicalDatastoreType store,
+ final YangInstanceIdentifier path) {
+ return getSubtransaction(store).exists(path);
+ }
+
+ @Override
+ public void close() {
+ closeSubtransactions();
+ }
+
+ @Override
+ protected T createTransaction(LogicalDatastoreType key) {
+ return (T) getTxFactory(key).newReadOnlyTransaction();
+ }
+
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.CheckedFuture;
+import java.util.Map;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+
+public class DOMBrokerReadWriteTransaction<T extends DOMStoreReadWriteTransaction>
+ extends DOMBrokerWriteOnlyTransaction<DOMStoreReadWriteTransaction> implements DOMDataReadWriteTransaction {
+ /**
+ * Creates new composite Transactions.
+ *
+ * @param identifier Identifier of transaction.
+ * @param storeTxFactories
+ */
+ protected DOMBrokerReadWriteTransaction(Object identifier, Map<LogicalDatastoreType, ? extends DOMStoreTransactionFactory> storeTxFactories, final AbstractDOMTransactionFactory<?> commitImpl) {
+ super(identifier, storeTxFactories, commitImpl);
+ }
+
+ @Override
+ public CheckedFuture<Optional<NormalizedNode<?,?>>, ReadFailedException> read(
+ final LogicalDatastoreType store, final YangInstanceIdentifier path) {
+ return getSubtransaction(store).read(path);
+ }
+
+ @Override
+ public CheckedFuture<Boolean, ReadFailedException> exists(
+ final LogicalDatastoreType store,
+ final YangInstanceIdentifier path) {
+ return getSubtransaction(store).exists(path);
+ }
+
+ @Override
+ protected DOMStoreReadWriteTransaction createTransaction(LogicalDatastoreType key) {
+ return getTxFactory(key).newReadWriteTransaction();
+ }
+
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
+import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class DOMBrokerTransactionChain extends AbstractDOMTransactionFactory<DOMStoreTransactionChain>
+ implements DOMTransactionChain {
+ private static enum State {
+ RUNNING,
+ CLOSING,
+ CLOSED,
+ FAILED,
+ }
+
+ private static final AtomicIntegerFieldUpdater<DOMBrokerTransactionChain> COUNTER_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(DOMBrokerTransactionChain.class, "counter");
+ private static final AtomicReferenceFieldUpdater<DOMBrokerTransactionChain, State> STATE_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(DOMBrokerTransactionChain.class, State.class, "state");
+ private static final Logger LOG = LoggerFactory.getLogger(DOMBrokerTransactionChain.class);
+ private final AtomicLong txNum = new AtomicLong();
+ private final AbstractDOMBroker broker;
+ private final TransactionChainListener listener;
+ private final long chainId;
+
+ private volatile State state = State.RUNNING;
+ private volatile int counter = 0;
+
+ /**
+ *
+ * @param chainId
+ * ID of transaction chain
+ * @param chains
+ * Backing {@link DOMStoreTransactionChain}s.
+ * @param listener
+ * Listener, which listens on transaction chain events.
+ * @throws NullPointerException
+ * If any of arguments is null.
+ */
+ public DOMBrokerTransactionChain(final long chainId,
+ final Map<LogicalDatastoreType, DOMStoreTransactionChain> chains,
+ AbstractDOMBroker broker, final TransactionChainListener listener) {
+ super(chains);
+ this.chainId = chainId;
+ this.broker = Preconditions.checkNotNull(broker);
+ this.listener = Preconditions.checkNotNull(listener);
+ }
+
+ private void checkNotFailed() {
+ Preconditions.checkState(state != State.FAILED, "Transaction chain has failed");
+ }
+
+ @Override
+ protected Object newTransactionIdentifier() {
+ return "DOM-CHAIN-" + chainId + "-" + txNum.getAndIncrement();
+ }
+
+ @Override
+ public CheckedFuture<Void, TransactionCommitFailedException> submit(
+ final DOMDataWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
+ checkNotFailed();
+ checkNotClosed();
+
+ final CheckedFuture<Void, TransactionCommitFailedException> ret = broker.submit(transaction, cohorts);
+
+ COUNTER_UPDATER.incrementAndGet(this);
+ Futures.addCallback(ret, new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(final Void result) {
+ transactionCompleted();
+ }
+
+ @Override
+ public void onFailure(final Throwable t) {
+ transactionFailed(transaction, t);
+ }
+ });
+
+ return ret;
+ }
+
+ @Override
+ public void close() {
+ final boolean success = STATE_UPDATER.compareAndSet(this, State.RUNNING, State.CLOSING);
+ if (!success) {
+ LOG.debug("Chain {} is no longer running", this);
+ return;
+ }
+
+ super.close();
+ for (DOMStoreTransactionChain subChain : getTxFactories().values()) {
+ subChain.close();
+ }
+
+ if (counter == 0) {
+ finishClose();
+ }
+ }
+
+ private void finishClose() {
+ state = State.CLOSED;
+ listener.onTransactionChainSuccessful(this);
+ }
+
+ private void transactionCompleted() {
+ if (COUNTER_UPDATER.decrementAndGet(this) == 0 && state == State.CLOSING) {
+ finishClose();
+ }
+ }
+
+ private void transactionFailed(final DOMDataWriteTransaction tx, final Throwable cause) {
+ state = State.FAILED;
+ LOG.debug("Transaction chain {}Â failed.", this, cause);
+ listener.onTransactionChainFailed(this, tx, cause);
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.databroker;
+
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.common.impl.service.AbstractDataTransaction;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DOMBrokerWriteOnlyTransaction<T extends DOMStoreWriteTransaction>
+ extends AbstractDOMBrokerTransaction<LogicalDatastoreType, T> implements DOMDataWriteTransaction {
+
+ private static final AtomicReferenceFieldUpdater<DOMBrokerWriteOnlyTransaction, AbstractDOMTransactionFactory> IMPL_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(DOMBrokerWriteOnlyTransaction.class, AbstractDOMTransactionFactory.class, "commitImpl");
+ @SuppressWarnings("rawtypes")
+ private static final AtomicReferenceFieldUpdater<DOMBrokerWriteOnlyTransaction, Future> FUTURE_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(DOMBrokerWriteOnlyTransaction.class, Future.class, "commitFuture");
+ private static final Logger LOG = LoggerFactory.getLogger(DOMBrokerWriteOnlyTransaction.class);
+ private static final Future<?> CANCELLED_FUTURE = Futures.immediateCancelledFuture();
+
+ /**
+ * Implementation of real commit. It also acts as an indication that
+ * the transaction is running -- which we flip atomically using
+ * {@link #IMPL_UPDATER}.
+ */
+ private volatile AbstractDOMTransactionFactory<?> commitImpl;
+
+ /**
+ * Future task of transaction commit. It starts off as null, but is
+ * set appropriately on {@link #submit()} and {@link #cancel()} via
+ * {@link AtomicReferenceFieldUpdater#lazySet(Object, Object)}.
+ *
+ * Lazy set is safe for use because it is only referenced to in the
+ * {@link #cancel()} slow path, where we will busy-wait for it. The
+ * fast path gets the benefit of a store-store barrier instead of the
+ * usual store-load barrier.
+ */
+ private volatile Future<?> commitFuture;
+
+ protected DOMBrokerWriteOnlyTransaction(final Object identifier,
+ Map storeTxFactories, final AbstractDOMTransactionFactory<?> commitImpl) {
+ super(identifier, storeTxFactories);
+ this.commitImpl = Preconditions.checkNotNull(commitImpl, "commitImpl must not be null.");
+ }
+
+ @Override
+ protected T createTransaction(LogicalDatastoreType key) {
+ // FIXME : Casting shouldn't be necessary here
+ return (T) getTxFactory(key).newWriteOnlyTransaction();
+ }
+
+ @Override
+ public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+ checkRunning(commitImpl);
+ getSubtransaction(store).write(path, data);
+ }
+
+ @Override
+ public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
+ checkRunning(commitImpl);
+ getSubtransaction(store).delete(path);
+ }
+
+ @Override
+ public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+ checkRunning(commitImpl);
+ getSubtransaction(store).merge(path, data);
+ }
+
+ @Override
+ public boolean cancel() {
+ final AbstractDOMTransactionFactory<?> impl = IMPL_UPDATER.getAndSet(this, null);
+ if (impl != null) {
+ LOG.trace("Transaction {} cancelled before submit", getIdentifier());
+ FUTURE_UPDATER.lazySet(this, CANCELLED_FUTURE);
+ closeSubtransactions();
+ return true;
+ }
+
+ // The transaction is in process of being submitted or cancelled. Busy-wait
+ // for the corresponding future.
+ Future<?> future;
+ do {
+ future = commitFuture;
+ } while (future == null);
+
+ return future.cancel(false);
+ }
+
+ @Deprecated
+ @Override
+ public ListenableFuture<RpcResult<TransactionStatus>> commit() {
+ return AbstractDataTransaction.convertToLegacyCommitFuture(submit());
+ }
+
+ @Override
+ public CheckedFuture<Void, TransactionCommitFailedException> submit() {
+ final AbstractDOMTransactionFactory<?> impl = IMPL_UPDATER.getAndSet(this, null);
+ checkRunning(impl);
+
+ final Collection<T> txns = getSubtransactions();
+ final Collection<DOMStoreThreePhaseCommitCohort> cohorts = new ArrayList<>(txns.size());
+
+ // FIXME: deal with errors thrown by backed (ready and submit can fail in theory)
+ for (DOMStoreWriteTransaction txn : txns) {
+ cohorts.add(txn.ready());
+ }
+
+ final CheckedFuture<Void, TransactionCommitFailedException> ret = impl.submit(this, cohorts);
+ FUTURE_UPDATER.lazySet(this, ret);
+ return ret;
+ }
+
+ private void checkRunning(final AbstractDOMTransactionFactory<?> impl) {
+ Preconditions.checkState(impl != null, "Transaction %s is no longer running", getIdentifier());
+ }
+
+}
*/
package org.opendaylight.controller.cluster.datastore;
-import akka.actor.ActorSelection;
import akka.dispatch.OnComplete;
import java.util.List;
+import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
* previous Tx's ready operations haven't completed yet.
*/
@Override
- protected Future<ActorSelection> sendFindPrimaryShardAsync(final String shardName) {
+ protected Future<PrimaryShardInfo> sendFindPrimaryShardAsync(final String shardName) {
// Check if there are any previous ready Futures, otherwise let the super class handle it.
if(previousReadyFutures.isEmpty()) {
return super.sendFindPrimaryShardAsync(shardName);
previousReadyFutures, getActorContext().getClientDispatcher());
// Add a callback for completion of the combined Futures.
- final Promise<ActorSelection> returnPromise = akka.dispatch.Futures.promise();
+ final Promise<PrimaryShardInfo> returnPromise = akka.dispatch.Futures.promise();
OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
@Override
public void onComplete(Throwable failure, Iterable<Object> notUsed) {
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import org.opendaylight.controller.cluster.databroker.AbstractDOMBroker;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
-import org.opendaylight.controller.md.sal.dom.broker.impl.AbstractDOMDataBroker;
import org.opendaylight.controller.md.sal.dom.broker.impl.TransactionCommitFailedExceptionMapper;
import org.opendaylight.controller.sal.core.spi.data.DOMStore;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.slf4j.LoggerFactory;
/**
- * Implementation of DOMDataCommitExecutor that coordinates transaction commits concurrently. The 3
+ * ConcurrentDOMDataBroker commits transactions concurrently. The 3
* commit phases (canCommit, preCommit, and commit) are performed serially and non-blocking
* (ie async) per transaction but multiple transaction commits can run concurrent.
*
* @author Thomas Pantelis
*/
-public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker {
+public class ConcurrentDOMDataBroker extends AbstractDOMBroker {
private static final Logger LOG = LoggerFactory.getLogger(ConcurrentDOMDataBroker.class);
private static final String CAN_COMMIT = "CAN_COMMIT";
private static final String PRE_COMMIT = "PRE_COMMIT";
import akka.actor.ActorSelection;
import akka.actor.PoisonPill;
import akka.dispatch.OnComplete;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
} else {
RegisterDataTreeChangeListenerReply reply = (RegisterDataTreeChangeListenerReply) result;
setListenerRegistrationActor(actorContext.actorSelection(
- reply.getListenerRegistrationPath().path()));
+ reply.getListenerRegistrationPath()));
}
}
}, actorContext.getClientDispatcher());
}
+
+ @VisibleForTesting
+ ActorSelection getListenerRegistrationActor() {
+ return listenerRegistrationActor;
+ }
+
+ @VisibleForTesting
+ ActorRef getDataChangeListenerActor() {
+ return dataChangeListenerActor;
+ }
}
if (chain != null) {
chain.close();
} else {
- LOG.warn("Closing non-existent transaction chain {}", transactionChainId);
+ LOG.debug("Closing non-existent transaction chain {}", transactionChainId);
}
}
import java.util.concurrent.atomic.AtomicLong;
import org.opendaylight.controller.cluster.datastore.compat.PreLithiumTransactionContextImpl;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
+import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator;
return ShardStrategyFactory.getStrategy(path).findShard(path);
}
- protected Future<ActorSelection> sendFindPrimaryShardAsync(String shardName) {
+ protected Future<PrimaryShardInfo> sendFindPrimaryShardAsync(String shardName) {
return actorContext.findPrimaryShardAsync(shardName);
}
private TransactionFutureCallback getOrCreateTxFutureCallback(String shardName) {
TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName);
if(txFutureCallback == null) {
- Future<ActorSelection> findPrimaryFuture = sendFindPrimaryShardAsync(shardName);
+ Future<PrimaryShardInfo> findPrimaryFuture = sendFindPrimaryShardAsync(shardName);
final TransactionFutureCallback newTxFutureCallback = new TransactionFutureCallback(this, shardName);
txFutureCallback = newTxFutureCallback;
txFutureCallbackMap.put(shardName, txFutureCallback);
- findPrimaryFuture.onComplete(new OnComplete<ActorSelection>() {
+ findPrimaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
@Override
- public void onComplete(Throwable failure, ActorSelection primaryShard) {
+ public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) {
if(failure != null) {
newTxFutureCallback.createTransactionContext(failure, null);
} else {
- newTxFutureCallback.setPrimaryShard(primaryShard);
+ newTxFutureCallback.setPrimaryShard(primaryShardInfo.getPrimaryShardActor());
}
}
}, actorContext.getClientDispatcher());
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore.messages;
+
+import akka.actor.ActorSelection;
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import javax.annotation.Nonnull;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
+
+/**
+ * Local message DTO that contains information about the primary shard.
+ *
+ * @author Thomas Pantelis
+ */
+public class PrimaryShardInfo {
+ private final ActorSelection primaryShardActor;
+ private final Optional<DataTree> localShardDataTree;
+
+ public PrimaryShardInfo(@Nonnull ActorSelection primaryShardActor, @Nonnull Optional<DataTree> localShardDataTree) {
+ this.primaryShardActor = Preconditions.checkNotNull(primaryShardActor);
+ this.localShardDataTree = Preconditions.checkNotNull(localShardDataTree);
+ }
+
+ /**
+ * Returns an ActorSelection representing the primary shard actor.
+ */
+ public @Nonnull ActorSelection getPrimaryShardActor() {
+ return primaryShardActor;
+ }
+
+ /**
+ * Returns an Optional whose value contains the primary shard's DataTree if the primary shard is local
+ * to the caller. Otherwise the Optional value is absent.
+ */
+ public @Nonnull Optional<DataTree> getLocalShardDataTree() {
+ return localShardDataTree;
+ }
+}
*/
package org.opendaylight.controller.cluster.datastore.messages;
+import akka.actor.ActorPath;
import akka.actor.ActorRef;
import com.google.common.base.Preconditions;
+
import java.io.Serializable;
/**
this.listenerRegistrationPath = Preconditions.checkNotNull(listenerRegistrationPath);
}
- public ActorRef getListenerRegistrationPath() {
- return listenerRegistrationPath;
+ public ActorPath getListenerRegistrationPath() {
+ return listenerRegistrationPath.path();
}
}
import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
+import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.reporting.MetricsReporter;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private Timeout transactionCommitOperationTimeout;
private Timeout shardInitializationTimeout;
private final Dispatchers dispatchers;
- private Cache<String, Future<ActorSelection>> primaryShardActorSelectionCache;
+ private Cache<String, Future<PrimaryShardInfo>> primaryShardInfoCache;
private volatile SchemaContext schemaContext;
private volatile boolean updated;
shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2));
- primaryShardActorSelectionCache = CacheBuilder.newBuilder()
+ primaryShardInfoCache = CacheBuilder.newBuilder()
.expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS)
.build();
}
return schemaContext;
}
- public Future<ActorSelection> findPrimaryShardAsync(final String shardName) {
- Future<ActorSelection> ret = primaryShardActorSelectionCache.getIfPresent(shardName);
+ public Future<PrimaryShardInfo> findPrimaryShardAsync(final String shardName) {
+ Future<PrimaryShardInfo> ret = primaryShardInfoCache.getIfPresent(shardName);
if(ret != null){
return ret;
}
Future<Object> future = executeOperationAsync(shardManager,
new FindPrimary(shardName, true), shardInitializationTimeout);
- return future.transform(new Mapper<Object, ActorSelection>() {
+ return future.transform(new Mapper<Object, PrimaryShardInfo>() {
@Override
- public ActorSelection checkedApply(Object response) throws Exception {
+ public PrimaryShardInfo checkedApply(Object response) throws Exception {
if(response instanceof PrimaryFound) {
PrimaryFound found = (PrimaryFound)response;
LOG.debug("Primary found {}", found.getPrimaryPath());
ActorSelection actorSelection = actorSystem.actorSelection(found.getPrimaryPath());
- primaryShardActorSelectionCache.put(shardName, Futures.successful(actorSelection));
- return actorSelection;
+ PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.<DataTree>absent());
+ primaryShardInfoCache.put(shardName, Futures.successful(info));
+ return info;
} else if(response instanceof NotInitializedException) {
throw (NotInitializedException)response;
} else if(response instanceof PrimaryNotFoundException) {
public void broadcast(final Object message){
for(final String shardName : configuration.getAllShardNames()){
- Future<ActorSelection> primaryFuture = findPrimaryShardAsync(shardName);
- primaryFuture.onComplete(new OnComplete<ActorSelection>() {
+ Future<PrimaryShardInfo> primaryFuture = findPrimaryShardAsync(shardName);
+ primaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
@Override
- public void onComplete(Throwable failure, ActorSelection primaryShard) {
+ public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) {
if(failure != null) {
LOG.warn("broadcast failed to send message {} to shard {}: {}",
message.getClass().getSimpleName(), shardName, failure);
} else {
- primaryShard.tell(message, ActorRef.noSender());
+ primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender());
}
}
}, getClientDispatcher());
}
@VisibleForTesting
- Cache<String, Future<ActorSelection>> getPrimaryShardActorSelectionCache() {
- return primaryShardActorSelectionCache;
+ Cache<String, Future<PrimaryShardInfo>> getPrimaryShardInfoCache() {
+ return primaryShardInfoCache;
}
}
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
+import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.messages.ReadData;
import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
return setupActorContextWithoutInitialCreateTransaction(actorSystem, DefaultShardStrategy.DEFAULT_SHARD);
}
+ protected Future<PrimaryShardInfo> primaryShardInfoReply(ActorSystem actorSystem, ActorRef actorRef) {
+ return Futures.successful(new PrimaryShardInfo(actorSystem.actorSelection(actorRef.path()),
+ Optional.<DataTree>absent()));
+ }
+
protected ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem, String shardName) {
ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
log.info("Created mock shard actor {}", actorRef);
doReturn(actorSystem.actorSelection(actorRef.path())).
when(mockActorContext).actorSelection(actorRef.path().toString());
- doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))).
+ doReturn(primaryShardInfoReply(actorSystem, actorRef)).
when(mockActorContext).findPrimaryShardAsync(eq(shardName));
doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString());
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
+import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
import org.opendaylight.controller.sal.core.spi.data.DOMStore;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
/**
* Unit tests for DOMConcurrentDataCommitCoordinator.
assertFailure(future, cause, mockCohort1, mockCohort2);
}
+
+ @Test
+ public void testCreateReadWriteTransaction(){
+ DOMStore domStore = mock(DOMStore.class);
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ domStore, LogicalDatastoreType.CONFIGURATION, domStore), futureExecutor);
+ dataBroker.newReadWriteTransaction();
+
+ verify(domStore, never()).newReadWriteTransaction();
+ }
+
+
+ @Test
+ public void testCreateWriteOnlyTransaction(){
+ DOMStore domStore = mock(DOMStore.class);
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ domStore, LogicalDatastoreType.CONFIGURATION, domStore), futureExecutor);
+ dataBroker.newWriteOnlyTransaction();
+
+ verify(domStore, never()).newWriteOnlyTransaction();
+ }
+
+ @Test
+ public void testCreateReadOnlyTransaction(){
+ DOMStore domStore = mock(DOMStore.class);
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ domStore, LogicalDatastoreType.CONFIGURATION, domStore), futureExecutor);
+ dataBroker.newReadOnlyTransaction();
+
+ verify(domStore, never()).newReadOnlyTransaction();
+ }
+
+ @Test
+ public void testLazySubTransactionCreationForReadWriteTransactions(){
+ DOMStore configDomStore = mock(DOMStore.class);
+ DOMStore operationalDomStore = mock(DOMStore.class);
+ DOMStoreReadWriteTransaction storeTxn = mock(DOMStoreReadWriteTransaction.class);
+
+ doReturn(storeTxn).when(operationalDomStore).newReadWriteTransaction();
+ doReturn(storeTxn).when(configDomStore).newReadWriteTransaction();
+
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor);
+ DOMDataReadWriteTransaction dataTxn = dataBroker.newReadWriteTransaction();
+
+ dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+ dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+ dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build());
+
+ verify(configDomStore, never()).newReadWriteTransaction();
+ verify(operationalDomStore, times(1)).newReadWriteTransaction();
+
+ dataTxn.put(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+
+ verify(configDomStore, times(1)).newReadWriteTransaction();
+ verify(operationalDomStore, times(1)).newReadWriteTransaction();
+
+ }
+
+ @Test
+ public void testLazySubTransactionCreationForWriteOnlyTransactions(){
+ DOMStore configDomStore = mock(DOMStore.class);
+ DOMStore operationalDomStore = mock(DOMStore.class);
+ DOMStoreWriteTransaction storeTxn = mock(DOMStoreWriteTransaction.class);
+
+ doReturn(storeTxn).when(operationalDomStore).newWriteOnlyTransaction();
+ doReturn(storeTxn).when(configDomStore).newWriteOnlyTransaction();
+
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor);
+ DOMDataWriteTransaction dataTxn = dataBroker.newWriteOnlyTransaction();
+
+ dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+ dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+
+ verify(configDomStore, never()).newWriteOnlyTransaction();
+ verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
+
+ dataTxn.put(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+
+ verify(configDomStore, times(1)).newWriteOnlyTransaction();
+ verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
+
+ }
+
+
+ @Test
+ public void testLazySubTransactionCreationForReadOnlyTransactions(){
+ DOMStore configDomStore = mock(DOMStore.class);
+ DOMStore operationalDomStore = mock(DOMStore.class);
+ DOMStoreReadTransaction storeTxn = mock(DOMStoreReadTransaction.class);
+
+ doReturn(storeTxn).when(operationalDomStore).newReadOnlyTransaction();
+ doReturn(storeTxn).when(configDomStore).newReadOnlyTransaction();
+
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor);
+ DOMDataReadOnlyTransaction dataTxn = dataBroker.newReadOnlyTransaction();
+
+ dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build());
+ dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build());
+
+ verify(configDomStore, never()).newReadOnlyTransaction();
+ verify(operationalDomStore, times(1)).newReadOnlyTransaction();
+
+ dataTxn.read(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.builder().build());
+
+ verify(configDomStore, times(1)).newReadOnlyTransaction();
+ verify(operationalDomStore, times(1)).newReadOnlyTransaction();
+
+ }
+
+ @Test
+ public void testSubmitWithOnlyOneSubTransaction() throws InterruptedException {
+ DOMStore configDomStore = mock(DOMStore.class);
+ DOMStore operationalDomStore = mock(DOMStore.class);
+ DOMStoreReadWriteTransaction mockStoreReadWriteTransaction = mock(DOMStoreReadWriteTransaction.class);
+ DOMStoreThreePhaseCommitCohort mockCohort = mock(DOMStoreThreePhaseCommitCohort.class);
+
+ doReturn(mockStoreReadWriteTransaction).when(operationalDomStore).newReadWriteTransaction();
+ doReturn(mockCohort).when(mockStoreReadWriteTransaction).ready();
+ doReturn(Futures.immediateFuture(false)).when(mockCohort).canCommit();
+ doReturn(Futures.immediateFuture(null)).when(mockCohort).abort();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList();
+
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor) {
+ @Override
+ public CheckedFuture<Void, TransactionCommitFailedException> submit(DOMDataWriteTransaction transaction, Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
+ commitCohorts.addAll(cohorts);
+ latch.countDown();
+ return super.submit(transaction, cohorts);
+ }
+ };
+ DOMDataReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
+
+ domDataReadWriteTransaction.delete(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build());
+
+ domDataReadWriteTransaction.submit();
+
+ latch.await(10, TimeUnit.SECONDS);
+
+ assertTrue(commitCohorts.size() == 1);
+ }
+
+ @Test
+ public void testSubmitWithOnlyTwoSubTransactions() throws InterruptedException {
+ DOMStore configDomStore = mock(DOMStore.class);
+ DOMStore operationalDomStore = mock(DOMStore.class);
+ DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
+ DOMStoreReadWriteTransaction configTransaction = mock(DOMStoreReadWriteTransaction.class);
+ DOMStoreThreePhaseCommitCohort mockCohortOperational = mock(DOMStoreThreePhaseCommitCohort.class);
+ DOMStoreThreePhaseCommitCohort mockCohortConfig = mock(DOMStoreThreePhaseCommitCohort.class);
+
+ doReturn(operationalTransaction).when(operationalDomStore).newReadWriteTransaction();
+ doReturn(configTransaction).when(configDomStore).newReadWriteTransaction();
+
+ doReturn(mockCohortOperational).when(operationalTransaction).ready();
+ doReturn(Futures.immediateFuture(false)).when(mockCohortOperational).canCommit();
+ doReturn(Futures.immediateFuture(null)).when(mockCohortOperational).abort();
+
+ doReturn(mockCohortConfig).when(configTransaction).ready();
+ doReturn(Futures.immediateFuture(false)).when(mockCohortConfig).canCommit();
+ doReturn(Futures.immediateFuture(null)).when(mockCohortConfig).abort();
+
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList();
+
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor) {
+ @Override
+ public CheckedFuture<Void, TransactionCommitFailedException> submit(DOMDataWriteTransaction transaction, Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
+ commitCohorts.addAll(cohorts);
+ latch.countDown();
+ return super.submit(transaction, cohorts);
+ }
+ };
+ DOMDataReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
+
+ domDataReadWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+ domDataReadWriteTransaction.merge(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+
+ domDataReadWriteTransaction.submit();
+
+ latch.await(10, TimeUnit.SECONDS);
+
+ assertTrue(commitCohorts.size() == 2);
+ }
+
+ @Test
+ public void testCreateTransactionChain(){
+ DOMStore domStore = mock(DOMStore.class);
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ domStore, LogicalDatastoreType.CONFIGURATION, domStore), futureExecutor);
+
+ dataBroker.createTransactionChain(mock(TransactionChainListener.class));
+
+ verify(domStore, times(2)).createTransactionChain();
+
+ }
+
+ @Test
+ public void testCreateTransactionOnChain(){
+ DOMStore domStore = mock(DOMStore.class);
+ ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL,
+ domStore, LogicalDatastoreType.CONFIGURATION, domStore), futureExecutor);
+
+ DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
+ DOMStoreTransactionChain mockChain = mock(DOMStoreTransactionChain.class);
+
+ doReturn(mockChain).when(domStore).createTransactionChain();
+ doReturn(operationalTransaction).when(mockChain).newWriteOnlyTransaction();
+
+ DOMTransactionChain transactionChain = dataBroker.createTransactionChain(mock(TransactionChainListener.class));
+
+ DOMDataWriteTransaction domDataWriteTransaction = transactionChain.newWriteOnlyTransaction();
+
+ verify(mockChain, never()).newWriteOnlyTransaction();
+
+ domDataWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.builder().build(), mock(NormalizedNode.class));
+ }
+
}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.DeadLetter;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.google.common.collect.ImmutableList;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.messages.DataTreeChanged;
+import org.opendaylight.controller.cluster.datastore.messages.DataTreeChangedReply;
+import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+
+public class DataTreeChangeListenerActorTest extends AbstractActorTest {
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testDataChangedWhenNotificationsAreEnabled(){
+ new JavaTestKit(getSystem()) {{
+ final DataTreeCandidate mockTreeCandidate = Mockito.mock(DataTreeCandidate.class);
+ final ImmutableList<DataTreeCandidate> mockCandidates = ImmutableList.of(mockTreeCandidate);
+ final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+ final Props props = DataTreeChangeListenerActor.props(mockListener);
+ final ActorRef subject = getSystem().actorOf(props, "testDataTreeChangedNotificationsEnabled");
+
+ // Let the DataChangeListener know that notifications should be enabled
+ subject.tell(new EnableNotification(true), getRef());
+
+ subject.tell(new DataTreeChanged(mockCandidates),
+ getRef());
+
+ expectMsgClass(DataTreeChangedReply.class);
+
+ Mockito.verify(mockListener).onDataTreeChanged(mockCandidates);
+ }};
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testDataChangedWhenNotificationsAreDisabled(){
+ new JavaTestKit(getSystem()) {{
+ final DataTreeCandidate mockTreeCandidate = Mockito.mock(DataTreeCandidate.class);
+ final ImmutableList<DataTreeCandidate> mockCandidates = ImmutableList.of(mockTreeCandidate);
+ final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+ final Props props = DataTreeChangeListenerActor.props(mockListener);
+ final ActorRef subject =
+ getSystem().actorOf(props, "testDataTreeChangedNotificationsDisabled");
+
+ subject.tell(new DataTreeChanged(mockCandidates),
+ getRef());
+
+ new Within(duration("1 seconds")) {
+ @Override
+ protected void run() {
+ expectNoMsg();
+
+ Mockito.verify(mockListener, Mockito.never()).onDataTreeChanged(
+ Matchers.anyCollectionOf(DataTreeCandidate.class));
+ }
+ };
+ }};
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testDataChangedWithNoSender(){
+ new JavaTestKit(getSystem()) {{
+ final DataTreeCandidate mockTreeCandidate = Mockito.mock(DataTreeCandidate.class);
+ final ImmutableList<DataTreeCandidate> mockCandidates = ImmutableList.of(mockTreeCandidate);
+ final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+ final Props props = DataTreeChangeListenerActor.props(mockListener);
+ final ActorRef subject = getSystem().actorOf(props, "testDataTreeChangedWithNoSender");
+
+ getSystem().eventStream().subscribe(getRef(), DeadLetter.class);
+
+ subject.tell(new DataTreeChanged(mockCandidates), ActorRef.noSender());
+
+ // Make sure no DataChangedReply is sent to DeadLetters.
+ while(true) {
+ DeadLetter deadLetter;
+ try {
+ deadLetter = expectMsgClass(duration("1 seconds"), DeadLetter.class);
+ } catch (AssertionError e) {
+ // Timed out - got no DeadLetter - this is good
+ break;
+ }
+
+ // We may get DeadLetters for other messages we don't care about.
+ Assert.assertFalse("Unexpected DataTreeChangedReply",
+ deadLetter.message() instanceof DataTreeChangedReply);
+ }
+ }};
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test
+ public void testDataChangedWithListenerRuntimeEx(){
+ new JavaTestKit(getSystem()) {{
+ final DataTreeCandidate mockTreeCandidate1 = Mockito.mock(DataTreeCandidate.class);
+ final ImmutableList<DataTreeCandidate> mockCandidates1 = ImmutableList.of(mockTreeCandidate1);
+ final DataTreeCandidate mockTreeCandidate2 = Mockito.mock(DataTreeCandidate.class);
+ final ImmutableList<DataTreeCandidate> mockCandidates2 = ImmutableList.of(mockTreeCandidate2);
+ final DataTreeCandidate mockTreeCandidate3 = Mockito.mock(DataTreeCandidate.class);
+ final ImmutableList<DataTreeCandidate> mockCandidates3 = ImmutableList.of(mockTreeCandidate3);
+
+ final DOMDataTreeChangeListener mockListener = Mockito.mock(DOMDataTreeChangeListener.class);
+ Mockito.doThrow(new RuntimeException("mock")).when(mockListener).onDataTreeChanged(mockCandidates2);
+
+ Props props = DataTreeChangeListenerActor.props(mockListener);
+ ActorRef subject = getSystem().actorOf(props, "testDataTreeChangedWithListenerRuntimeEx");
+
+ // Let the DataChangeListener know that notifications should be enabled
+ subject.tell(new EnableNotification(true), getRef());
+
+ subject.tell(new DataTreeChanged(mockCandidates1),getRef());
+ expectMsgClass(DataTreeChangedReply.class);
+
+ subject.tell(new DataTreeChanged(mockCandidates2),getRef());
+ expectMsgClass(DataTreeChangedReply.class);
+
+ subject.tell(new DataTreeChanged(mockCandidates3),getRef());
+ expectMsgClass(DataTreeChangedReply.class);
+
+ Mockito.verify(mockListener).onDataTreeChanged(mockCandidates1);
+ Mockito.verify(mockListener).onDataTreeChanged(mockCandidates2);
+ Mockito.verify(mockListener).onDataTreeChanged(mockCandidates3);
+ }};
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.actor.Terminated;
+import akka.dispatch.ExecutionContexts;
+import akka.dispatch.Futures;
+import akka.testkit.JavaTestKit;
+import akka.util.Timeout;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Uninterruptibles;
+import java.util.concurrent.TimeUnit;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeChangeListenerRegistration;
+import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
+import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
+import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply;
+import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
+import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import scala.concurrent.ExecutionContextExecutor;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
+public class DataTreeChangeListenerProxyTest extends AbstractActorTest {
+ @SuppressWarnings("unchecked")
+ private final DOMDataTreeChangeListener mockListener = mock(DOMDataTreeChangeListener.class);
+
+ @Test(timeout=10000)
+ public void testSuccessfulRegistration() {
+ new JavaTestKit(getSystem()) {{
+ ActorContext actorContext = new ActorContext(getSystem(), getRef(),
+ mock(ClusterWrapper.class), mock(Configuration.class));
+
+ final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+ new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+ final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+ new Thread() {
+ @Override
+ public void run() {
+ proxy.init("shard-1", path);
+ }
+
+ }.start();
+
+ FiniteDuration timeout = duration("5 seconds");
+ FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class);
+ Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName());
+
+ reply(new LocalShardFound(getRef()));
+
+ RegisterDataTreeChangeListener registerMsg = expectMsgClass(timeout, RegisterDataTreeChangeListener.class);
+ Assert.assertEquals("getPath", path, registerMsg.getPath());
+
+ reply(new RegisterDataTreeChangeListenerReply(getRef()));
+
+
+ for(int i = 0; (i < 20 * 5) && proxy.getListenerRegistrationActor() == null; i++) {
+ Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
+ }
+
+ Assert.assertEquals("getListenerRegistrationActor", getSystem().actorSelection(getRef().path()),
+ proxy.getListenerRegistrationActor());
+
+ watch(proxy.getDataChangeListenerActor());
+
+ proxy.close();
+
+ // The listener registration actor should get a Close message
+ expectMsgClass(timeout, CloseDataTreeChangeListenerRegistration.class);
+
+ // The DataChangeListener actor should be terminated
+ expectMsgClass(timeout, Terminated.class);
+
+ proxy.close();
+
+ expectNoMsg();
+ }};
+ }
+
+ @Test(timeout=10000)
+ public void testLocalShardNotFound() {
+ new JavaTestKit(getSystem()) {{
+ ActorContext actorContext = new ActorContext(getSystem(), getRef(),
+ mock(ClusterWrapper.class), mock(Configuration.class));
+
+ final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+ new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+ final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+ new Thread() {
+ @Override
+ public void run() {
+ proxy.init("shard-1", path);
+ }
+
+ }.start();
+
+ FiniteDuration timeout = duration("5 seconds");
+ FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class);
+ Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName());
+
+ reply(new LocalShardNotFound("shard-1"));
+
+ expectNoMsg(duration("1 seconds"));
+ }};
+ }
+
+ @Test(timeout=10000)
+ public void testLocalShardNotInitialized() {
+ new JavaTestKit(getSystem()) {{
+ ActorContext actorContext = new ActorContext(getSystem(), getRef(),
+ mock(ClusterWrapper.class), mock(Configuration.class));
+
+ final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+ new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+ final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+ new Thread() {
+ @Override
+ public void run() {
+ proxy.init("shard-1", path);
+ }
+
+ }.start();
+
+ FiniteDuration timeout = duration("5 seconds");
+ FindLocalShard findLocalShard = expectMsgClass(timeout, FindLocalShard.class);
+ Assert.assertEquals("getShardName", "shard-1", findLocalShard.getShardName());
+
+ reply(new NotInitializedException("not initialized"));
+
+ new Within(duration("1 seconds")) {
+ @Override
+ protected void run() {
+ expectNoMsg();
+ }
+ };
+ }};
+ }
+
+ @Test
+ public void testFailedRegistration() {
+ new JavaTestKit(getSystem()) {{
+ ActorSystem mockActorSystem = mock(ActorSystem.class);
+
+ ActorRef mockActor = getSystem().actorOf(Props.create(DoNothingActor.class),
+ "testFailedRegistration");
+ doReturn(mockActor).when(mockActorSystem).actorOf(any(Props.class));
+ ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(
+ MoreExecutors.sameThreadExecutor());
+
+
+ ActorContext actorContext = mock(ActorContext.class);
+ final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
+
+ doReturn(executor).when(actorContext).getClientDispatcher();
+ doReturn(mockActorSystem).when(actorContext).getActorSystem();
+
+ String shardName = "shard-1";
+ final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+ new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+ doReturn(duration("5 seconds")).when(actorContext).getOperationDuration();
+ doReturn(Futures.successful(getRef())).when(actorContext).findLocalShardAsync(eq(shardName));
+ doReturn(Futures.failed(new RuntimeException("mock"))).
+ when(actorContext).executeOperationAsync(any(ActorRef.class),
+ any(Object.class), any(Timeout.class));
+ doReturn(mock(DatastoreContext.class)).when(actorContext).getDatastoreContext();
+
+ proxy.init("shard-1", path);
+
+ Assert.assertEquals("getListenerRegistrationActor", null,
+ proxy.getListenerRegistrationActor());
+ }};
+ }
+
+ @Test
+ public void testCloseBeforeRegistration() {
+ new JavaTestKit(getSystem()) {{
+ ActorContext actorContext = mock(ActorContext.class);
+
+ String shardName = "shard-1";
+
+ doReturn(DatastoreContext.newBuilder().build()).when(actorContext).getDatastoreContext();
+ doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(actorContext).getClientDispatcher();
+ doReturn(getSystem()).when(actorContext).getActorSystem();
+ doReturn(Dispatchers.DEFAULT_DISPATCHER_PATH).when(actorContext).getNotificationDispatcherPath();
+ doReturn(getSystem().actorSelection(getRef().path())).
+ when(actorContext).actorSelection(getRef().path());
+ doReturn(duration("5 seconds")).when(actorContext).getOperationDuration();
+ doReturn(Futures.successful(getRef())).when(actorContext).findLocalShardAsync(eq(shardName));
+
+ final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
+ new DataTreeChangeListenerProxy<>(actorContext, mockListener);
+
+
+ Answer<Future<Object>> answer = new Answer<Future<Object>>() {
+ @Override
+ public Future<Object> answer(InvocationOnMock invocation) {
+ proxy.close();
+ return Futures.successful((Object)new RegisterDataTreeChangeListenerReply(getRef()));
+ }
+ };
+
+ doAnswer(answer).when(actorContext).executeOperationAsync(any(ActorRef.class),
+ any(Object.class), any(Timeout.class));
+
+ proxy.init(shardName, YangInstanceIdentifier.of(TestModel.TEST_QNAME));
+
+ expectMsgClass(duration("5 seconds"), CloseDataTreeChangeListenerRegistration.class);
+
+ Assert.assertEquals("getListenerRegistrationActor", null,
+ proxy.getListenerRegistrationActor());
+ }};
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeChangeListenerRegistration;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeChangeListenerRegistrationReply;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+
+public class DataTreeChangeListenerRegistrationActorTest extends AbstractActorTest {
+ private static final InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
+
+ static {
+ store.onGlobalContextUpdated(TestModel.createTestContext());
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Test
+ public void testOnReceiveCloseListenerRegistration() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ListenerRegistration mockListenerReg = Mockito.mock(ListenerRegistration.class);
+ final Props props = DataTreeChangeListenerRegistrationActor.props(mockListenerReg);
+ final ActorRef subject = getSystem().actorOf(props, "testCloseListenerRegistration");
+
+ subject.tell(CloseDataTreeChangeListenerRegistration.getInstance(), getRef());
+
+ expectMsgClass(duration("1 second"), CloseDataTreeChangeListenerRegistrationReply.class);
+
+ Mockito.verify(mockListenerReg).close();
+ }};
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import java.util.Arrays;
+import java.util.Collection;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.messages.DataTreeChanged;
+import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+
+public class ForwardingDataTreeChangeListenerTest extends AbstractActorTest {
+
+ @Test
+ public void testOnDataChanged() throws Exception {
+ final Props props = Props.create(MessageCollectorActor.class);
+ final ActorRef actorRef = getSystem().actorOf(props);
+
+ ForwardingDataTreeChangeListener forwardingListener = new ForwardingDataTreeChangeListener(
+ getSystem().actorSelection(actorRef.path()));
+
+ Collection<DataTreeCandidate> expected = Arrays.asList(Mockito.mock(DataTreeCandidate.class));
+ forwardingListener.onDataTreeChanged(expected);
+
+ DataTreeChanged actual = MessageCollectorActor.expectFirstMatching(actorRef, DataTreeChanged.class);
+ Assert.assertSame(expected, actual.getChanges());
+ }
+}
import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
+import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
}};
}
+ @Test
+ public void testRegisterDataTreeChangeListener() throws Exception {
+ new ShardTestKit(getSystem()) {{
+ TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
+ newShardProps(), "testRegisterDataTreeChangeListener");
+
+ waitUntilLeader(shard);
+
+ shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
+
+ MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
+ ActorRef dclActor = getSystem().actorOf(DataTreeChangeListenerActor.props(listener),
+ "testRegisterDataTreeChangeListener-DataTreeChangeListener");
+
+ shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor), getRef());
+
+ RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
+ RegisterDataTreeChangeListenerReply.class);
+ String replyPath = reply.getListenerRegistrationPath().toString();
+ assertTrue("Incorrect reply path: " + replyPath, replyPath.matches(
+ "akka:\\/\\/test\\/user\\/testRegisterDataTreeChangeListener\\/\\$.*"));
+
+ YangInstanceIdentifier path = TestModel.TEST_PATH;
+ writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ listener.waitForChangeEvents();
+
+ dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ }};
+ }
+
+ @SuppressWarnings("serial")
+ @Test
+ public void testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration() throws Exception {
+ new ShardTestKit(getSystem()) {{
+ final CountDownLatch onFirstElectionTimeout = new CountDownLatch(1);
+ final CountDownLatch onChangeListenerRegistered = new CountDownLatch(1);
+ Creator<Shard> creator = new Creator<Shard>() {
+ boolean firstElectionTimeout = true;
+
+ @Override
+ public Shard create() throws Exception {
+ return new Shard(shardID, Collections.<String,String>emptyMap(),
+ dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) {
+ @Override
+ public void onReceiveCommand(final Object message) throws Exception {
+ if(message instanceof ElectionTimeout && firstElectionTimeout) {
+ firstElectionTimeout = false;
+ final ActorRef self = getSelf();
+ new Thread() {
+ @Override
+ public void run() {
+ Uninterruptibles.awaitUninterruptibly(
+ onChangeListenerRegistered, 5, TimeUnit.SECONDS);
+ self.tell(message, self);
+ }
+ }.start();
+
+ onFirstElectionTimeout.countDown();
+ } else {
+ super.onReceiveCommand(message);
+ }
+ }
+ };
+ }
+ };
+
+ MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
+ ActorRef dclActor = getSystem().actorOf(DataTreeChangeListenerActor.props(listener),
+ "testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration-DataChangeListener");
+
+ TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
+ Props.create(new DelegatingShardCreator(creator)),
+ "testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration");
+
+ YangInstanceIdentifier path = TestModel.TEST_PATH;
+ writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ assertEquals("Got first ElectionTimeout", true,
+ onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
+
+ shard.tell(new RegisterDataTreeChangeListener(path, dclActor), getRef());
+ RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeChangeListenerReply.class);
+ assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath());
+
+ shard.tell(new FindLeader(), getRef());
+ FindLeaderReply findLeadeReply =
+ expectMsgClass(duration("5 seconds"), FindLeaderReply.class);
+ assertNull("Expected the shard not to be the leader", findLeadeReply.getLeaderActor());
+
+ writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ onChangeListenerRegistered.countDown();
+
+ // TODO: investigate why we do not receive data chage events
+ listener.waitForChangeEvents();
+
+ dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ }};
+ }
+
@Test
public void testCreateTransaction(){
new ShardTestKit(getSystem()) {{
if (exToThrow instanceof PrimaryNotFoundException) {
doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
} else {
- doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
+ doReturn(primaryShardInfoReply(getSystem(), actorRef)).
when(mockActorContext).findPrimaryShardAsync(anyString());
}
doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext).
actorSelection(actorRef.path().toString());
- doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))).
+ doReturn(primaryShardInfoReply(getSystem(), actorRef)).
when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
when(mockActorContext).actorSelection(shardActorRef.path().toString());
if(shardFound) {
- doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))).
+ doReturn(primaryShardInfoReply(actorSystem, shardActorRef)).
when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
} else {
doReturn(Futures.failed(new PrimaryNotFoundException("test")))
doReturn(getSystem().actorSelection(shardActorRef.path())).
when(mockActorContext).actorSelection(shardActorRef.path().toString());
- doReturn(Futures.successful(getSystem().actorSelection(shardActorRef.path()))).
+ doReturn(primaryShardInfoReply(getSystem(), shardActorRef)).
when(mockActorContext).findPrimaryShardAsync(eq(shardName));
doReturn(true).when(mockActorContext).isPathLocal(shardActorRef.path().toString());
import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
+import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Await;
DatastoreContext dataStoreContext = DatastoreContext.newBuilder().dataStoreType("config").
shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
+ final String expPrimaryPath = "akka://test-system/find-primary-shard";
ActorContext actorContext =
new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext) {
@Override
protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout) {
- return Futures.successful((Object) new PrimaryFound("akka://test-system/test"));
+ return Futures.successful((Object) new PrimaryFound(expPrimaryPath));
}
};
- Future<ActorSelection> foobar = actorContext.findPrimaryShardAsync("foobar");
- ActorSelection actual = Await.result(foobar, Duration.apply(5000, TimeUnit.MILLISECONDS));
+ Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
+ PrimaryShardInfo actual = Await.result(foobar, Duration.apply(5000, TimeUnit.MILLISECONDS));
assertNotNull(actual);
+ assertEquals("LocalShardDataTree present", false, actual.getLocalShardDataTree().isPresent());
+ assertTrue("Unexpected PrimaryShardActor path " + actual.getPrimaryShardActor().path(),
+ expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
- Future<ActorSelection> cached = actorContext.getPrimaryShardActorSelectionCache().getIfPresent("foobar");
+ Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
- ActorSelection cachedSelection = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
+ PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
- assertEquals(cachedSelection, actual);
+ assertEquals(cachedInfo, actual);
// Wait for 200 Milliseconds. The cached entry should have been removed.
Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
- cached = actorContext.getPrimaryShardActorSelectionCache().getIfPresent("foobar");
+ cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
assertNull(cached);
};
- Future<ActorSelection> foobar = actorContext.findPrimaryShardAsync("foobar");
+ Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
try {
Await.result(foobar, Duration.apply(100, TimeUnit.MILLISECONDS));
}
- Future<ActorSelection> cached = actorContext.getPrimaryShardActorSelectionCache().getIfPresent("foobar");
+ Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
assertNull(cached);
}
};
- Future<ActorSelection> foobar = actorContext.findPrimaryShardAsync("foobar");
+ Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
try {
Await.result(foobar, Duration.apply(100, TimeUnit.MILLISECONDS));
}
- Future<ActorSelection> cached = actorContext.getPrimaryShardActorSelectionCache().getIfPresent("foobar");
+ Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
assertNull(cached);
}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore.utils;
+
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+
+import javax.annotation.Nonnull;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class MockDataTreeChangeListener implements DOMDataTreeChangeListener {
+
+ private final List<Collection<DataTreeCandidate>> changeList =
+ Collections.synchronizedList(Lists.<Collection<DataTreeCandidate>>newArrayList());
+
+ private volatile CountDownLatch changeLatch;
+ private int expChangeEventCount;
+
+ public MockDataTreeChangeListener(int expChangeEventCount) {
+ reset(expChangeEventCount);
+ }
+
+ public void reset(int expChangeEventCount) {
+ changeLatch = new CountDownLatch(expChangeEventCount);
+ this.expChangeEventCount = expChangeEventCount;
+ changeList.clear();
+ }
+
+ @Override
+ public void onDataTreeChanged(@Nonnull final Collection<DataTreeCandidate> changes) {
+ changeList.add(changes);
+ changeLatch.countDown();
+ }
+
+ public void waitForChangeEvents() {
+ boolean done = Uninterruptibles.awaitUninterruptibly(changeLatch, 5, TimeUnit.SECONDS);
+ if(!done) {
+ fail(String.format("Missing change notifications. Expected: %d. Actual: %d",
+ expChangeEventCount, (expChangeEventCount - changeLatch.getCount())));
+ }
+ }
+
+ public void expectNoMoreChanges(String assertMsg) {
+ Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+ assertEquals(assertMsg, expChangeEventCount, changeList.size());
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.md.sal.dom.broker.impl;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertFalse;
+import static junit.framework.Assert.assertNotNull;
+import static junit.framework.Assert.assertTrue;
+import static org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION;
+import static org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.OPERATIONAL;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ForwardingExecutorService;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitDeadlockException;
+import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataBrokerExtension;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeService;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeIdentifier;
+import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
+import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
+import org.opendaylight.controller.md.sal.dom.store.impl.TestModel;
+import org.opendaylight.controller.sal.core.spi.data.DOMStore;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.util.concurrent.DeadlockDetectingListeningExecutorService;
+import org.opendaylight.yangtools.util.concurrent.SpecialExecutors;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
+import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
+import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
+import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+
+public class DOMDataTreeListenerTest {
+
+ private SchemaContext schemaContext;
+ private AbstractDOMDataBroker domBroker;
+ private ListeningExecutorService executor;
+ private ExecutorService futureExecutor;
+ private CommitExecutorService commitExecutor;
+
+ private static final DataContainerChild<?, ?> OUTER_LIST = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
+ .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1))
+ .build();
+
+ private static final DataContainerChild<?, ?> OUTER_LIST_2 = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
+ .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2))
+ .build();
+
+ private static final NormalizedNode<?, ?> TEST_CONTAINER = Builders.containerBuilder()
+ .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME))
+ .withChild(OUTER_LIST)
+ .build();
+
+ private static final NormalizedNode<?, ?> TEST_CONTAINER_2 = Builders.containerBuilder()
+ .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME))
+ .withChild(OUTER_LIST_2)
+ .build();
+
+ private static DOMDataTreeIdentifier ROOT_DATA_TREE_ID = new DOMDataTreeIdentifier(
+ LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
+
+ private static DOMDataTreeIdentifier OUTER_LIST_DATA_TREE_ID = new DOMDataTreeIdentifier(
+ LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_LIST_PATH);
+
+ @Before
+ public void setupStore() {
+ InMemoryDOMDataStore operStore = new InMemoryDOMDataStore("OPER",
+ MoreExecutors.newDirectExecutorService());
+ InMemoryDOMDataStore configStore = new InMemoryDOMDataStore("CFG",
+ MoreExecutors.newDirectExecutorService());
+ schemaContext = TestModel.createTestContext();
+
+ operStore.onGlobalContextUpdated(schemaContext);
+ configStore.onGlobalContextUpdated(schemaContext);
+
+ ImmutableMap<LogicalDatastoreType, DOMStore> stores = ImmutableMap.<LogicalDatastoreType, DOMStore>builder() //
+ .put(CONFIGURATION, configStore) //
+ .put(OPERATIONAL, operStore) //
+ .build();
+
+ commitExecutor = new CommitExecutorService(Executors.newSingleThreadExecutor());
+ futureExecutor = SpecialExecutors.newBlockingBoundedCachedThreadPool(1, 5, "FCB");
+ executor = new DeadlockDetectingListeningExecutorService(commitExecutor,
+ TransactionCommitDeadlockException.DEADLOCK_EXCEPTION_SUPPLIER, futureExecutor);
+ domBroker = new SerializedDOMDataBroker(stores, executor);
+ }
+
+ @After
+ public void tearDown() {
+ if (executor != null) {
+ executor.shutdownNow();
+ }
+
+ if (futureExecutor != null) {
+ futureExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void writeContainerEmptyTreeTest() throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ DOMDataTreeChangeService dataTreeChangeService = getDOMDataTreeChangeService();
+ assertNotNull("DOMDataTreeChangeService not found, cannot continue with test!",
+ dataTreeChangeService);
+
+ final TestDataTreeListener listener = new TestDataTreeListener(latch);
+ final ListenerRegistration<TestDataTreeListener> listenerReg =
+ dataTreeChangeService.registerDataTreeChangeListener(ROOT_DATA_TREE_ID, listener);
+
+ final DOMDataWriteTransaction writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER);
+ writeTx.submit();
+
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(1, listener.getReceivedChanges().size());
+ final Collection<DataTreeCandidate> changes = listener.getReceivedChanges().get(0);
+ assertEquals(1, changes.size());
+
+ DataTreeCandidate candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ DataTreeCandidateNode candidateRoot = candidate.getRootNode();
+ checkChange(null, TEST_CONTAINER, ModificationType.WRITE, candidateRoot);
+ listenerReg.close();
+ }
+
+ @Test
+ public void replaceContainerContainerInTreeTest() throws InterruptedException, TransactionCommitFailedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ DOMDataTreeChangeService dataTreeChangeService = getDOMDataTreeChangeService();
+ assertNotNull("DOMDataTreeChangeService not found, cannot continue with test!",
+ dataTreeChangeService);
+
+ DOMDataWriteTransaction writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER);
+ writeTx.submit().checkedGet();
+
+ final TestDataTreeListener listener = new TestDataTreeListener(latch);
+ final ListenerRegistration<TestDataTreeListener> listenerReg =
+ dataTreeChangeService.registerDataTreeChangeListener(ROOT_DATA_TREE_ID, listener);
+ writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER_2);
+ writeTx.submit();
+
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(2, listener.getReceivedChanges().size());
+ Collection<DataTreeCandidate> changes = listener.getReceivedChanges().get(0);
+ assertEquals(1, changes.size());
+
+ DataTreeCandidate candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ DataTreeCandidateNode candidateRoot = candidate.getRootNode();
+ checkChange(null, TEST_CONTAINER, ModificationType.WRITE, candidateRoot);
+
+ changes = listener.getReceivedChanges().get(1);
+ assertEquals(1, changes.size());
+
+ candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ candidateRoot = candidate.getRootNode();
+ checkChange(TEST_CONTAINER, TEST_CONTAINER_2, ModificationType.WRITE, candidateRoot);
+ listenerReg.close();
+ }
+
+ @Test
+ public void deleteContainerContainerInTreeTest() throws InterruptedException, TransactionCommitFailedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ DOMDataTreeChangeService dataTreeChangeService = getDOMDataTreeChangeService();
+ assertNotNull("DOMDataTreeChangeService not found, cannot continue with test!",
+ dataTreeChangeService);
+
+ DOMDataWriteTransaction writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER);
+ writeTx.submit().checkedGet();
+
+ final TestDataTreeListener listener = new TestDataTreeListener(latch);
+ final ListenerRegistration<TestDataTreeListener> listenerReg =
+ dataTreeChangeService.registerDataTreeChangeListener(ROOT_DATA_TREE_ID, listener);
+
+ writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.delete(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
+ writeTx.submit();
+
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(2, listener.getReceivedChanges().size());
+ Collection<DataTreeCandidate> changes = listener.getReceivedChanges().get(0);
+ assertEquals(1, changes.size());
+
+ DataTreeCandidate candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ DataTreeCandidateNode candidateRoot = candidate.getRootNode();
+ checkChange(null, TEST_CONTAINER, ModificationType.WRITE, candidateRoot);
+
+ changes = listener.getReceivedChanges().get(1);
+ assertEquals(1, changes.size());
+
+ candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ candidateRoot = candidate.getRootNode();
+ checkChange(TEST_CONTAINER, null, ModificationType.DELETE, candidateRoot);
+ listenerReg.close();
+ }
+
+ @Test
+ public void replaceChildListContainerInTreeTest() throws InterruptedException, TransactionCommitFailedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ DOMDataTreeChangeService dataTreeChangeService = getDOMDataTreeChangeService();
+ assertNotNull("DOMDataTreeChangeService not found, cannot continue with test!",
+ dataTreeChangeService);
+
+ DOMDataWriteTransaction writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER);
+ writeTx.submit().checkedGet();
+
+ final TestDataTreeListener listener = new TestDataTreeListener(latch);
+ final ListenerRegistration<TestDataTreeListener> listenerReg =
+ dataTreeChangeService.registerDataTreeChangeListener(ROOT_DATA_TREE_ID, listener);
+
+ writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_LIST_PATH, OUTER_LIST_2);
+ writeTx.submit();
+
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(2, listener.getReceivedChanges().size());
+ Collection<DataTreeCandidate> changes = listener.getReceivedChanges().get(0);
+ assertEquals(1, changes.size());
+
+ DataTreeCandidate candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ DataTreeCandidateNode candidateRoot = candidate.getRootNode();
+ checkChange(null, TEST_CONTAINER, ModificationType.WRITE, candidateRoot);
+
+ changes = listener.getReceivedChanges().get(1);
+ assertEquals(1, changes.size());
+
+ candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ candidateRoot = candidate.getRootNode();
+ checkChange(TEST_CONTAINER, TEST_CONTAINER_2, ModificationType.SUBTREE_MODIFIED, candidateRoot);
+ final DataTreeCandidateNode modifiedChild = candidateRoot.getModifiedChild(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.OUTER_LIST_QNAME));
+ assertNotNull(modifiedChild);
+ checkChange(OUTER_LIST, OUTER_LIST_2, ModificationType.WRITE, modifiedChild);
+ listenerReg.close();
+ }
+
+ @Test
+ public void rootModificationChildListenerTest() throws InterruptedException, TransactionCommitFailedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ DOMDataTreeChangeService dataTreeChangeService = getDOMDataTreeChangeService();
+ assertNotNull("DOMDataTreeChangeService not found, cannot continue with test!",
+ dataTreeChangeService);
+
+ DOMDataWriteTransaction writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER);
+ writeTx.submit().checkedGet();
+
+ final TestDataTreeListener listener = new TestDataTreeListener(latch);
+ final ListenerRegistration<TestDataTreeListener> listenerReg =
+ dataTreeChangeService.registerDataTreeChangeListener(OUTER_LIST_DATA_TREE_ID, listener);
+
+ writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER_2);
+ writeTx.submit().checkedGet();
+
+ latch.await(1, TimeUnit.SECONDS);
+
+ assertEquals(2, listener.getReceivedChanges().size());
+ Collection<DataTreeCandidate> changes = listener.getReceivedChanges().get(0);
+ assertEquals(1, changes.size());
+
+ DataTreeCandidate candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ DataTreeCandidateNode candidateRoot = candidate.getRootNode();
+ checkChange(null, OUTER_LIST, ModificationType.WRITE, candidateRoot);
+
+ changes = listener.getReceivedChanges().get(1);
+ assertEquals(1, changes.size());
+
+ candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ candidateRoot = candidate.getRootNode();
+ checkChange(OUTER_LIST, OUTER_LIST_2, ModificationType.WRITE, candidateRoot);
+ listenerReg.close();
+ }
+
+ @Test
+ public void listEntryChangeNonRootRegistrationTest() throws InterruptedException, TransactionCommitFailedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ DOMDataTreeChangeService dataTreeChangeService = getDOMDataTreeChangeService();
+ assertNotNull("DOMDataTreeChangeService not found, cannot continue with test!",
+ dataTreeChangeService);
+
+ DOMDataWriteTransaction writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH, TEST_CONTAINER);
+ writeTx.submit().checkedGet();
+
+ final TestDataTreeListener listener = new TestDataTreeListener(latch);
+ final ListenerRegistration<TestDataTreeListener> listenerReg =
+ dataTreeChangeService.registerDataTreeChangeListener(OUTER_LIST_DATA_TREE_ID, listener);
+
+ final YangInstanceIdentifier.NodeIdentifierWithPredicates outerListEntryId1 =
+ new YangInstanceIdentifier.NodeIdentifierWithPredicates(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1);
+ final YangInstanceIdentifier.NodeIdentifierWithPredicates outerListEntryId2 =
+ new YangInstanceIdentifier.NodeIdentifierWithPredicates(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2);
+ final YangInstanceIdentifier.NodeIdentifierWithPredicates outerListEntryId3 =
+ new YangInstanceIdentifier.NodeIdentifierWithPredicates(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 3);
+
+ final MapEntryNode outerListEntry1 = ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1);
+ final MapEntryNode outerListEntry2 = ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2);
+ final MapEntryNode outerListEntry3 = ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 3);
+
+ final MapNode listAfter = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
+ .withChild(outerListEntry2)
+ .withChild(outerListEntry3)
+ .build();
+
+ writeTx = domBroker.newWriteOnlyTransaction();
+ writeTx.delete(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_LIST_PATH.node(outerListEntryId1));
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_LIST_PATH.node(outerListEntryId2),
+ outerListEntry2);
+ writeTx.put(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_LIST_PATH.node(outerListEntryId3),
+ outerListEntry3);
+ writeTx.submit();
+
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(2, listener.getReceivedChanges().size());
+ Collection<DataTreeCandidate> changes = listener.getReceivedChanges().get(0);
+ assertEquals(1, changes.size());
+
+ DataTreeCandidate candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ DataTreeCandidateNode candidateRoot = candidate.getRootNode();
+ checkChange(null, OUTER_LIST, ModificationType.WRITE, candidateRoot);
+
+ changes = listener.getReceivedChanges().get(1);
+ assertEquals(1, changes.size());
+
+ candidate = changes.iterator().next();
+ assertNotNull(candidate);
+ candidateRoot = candidate.getRootNode();
+ checkChange(OUTER_LIST, listAfter, ModificationType.SUBTREE_MODIFIED, candidateRoot);
+ final DataTreeCandidateNode entry1Canditate = candidateRoot.getModifiedChild(outerListEntryId1);
+ checkChange(outerListEntry1, null, ModificationType.DELETE, entry1Canditate);
+ final DataTreeCandidateNode entry2Canditate = candidateRoot.getModifiedChild(outerListEntryId2);
+ checkChange(null, outerListEntry2, ModificationType.WRITE, entry2Canditate);
+ final DataTreeCandidateNode entry3Canditate = candidateRoot.getModifiedChild(outerListEntryId3);
+ checkChange(null, outerListEntry3, ModificationType.WRITE, entry3Canditate);
+ listenerReg.close();
+ }
+
+ private static void checkChange(NormalizedNode<?, ?> expectedBefore,
+ NormalizedNode<?, ?> expectedAfter,
+ ModificationType expectedMod,
+ DataTreeCandidateNode candidateNode) {
+ if (expectedBefore != null) {
+ assertTrue(candidateNode.getDataBefore().isPresent());
+ assertEquals(expectedBefore, candidateNode.getDataBefore().get());
+ } else {
+ assertFalse(candidateNode.getDataBefore().isPresent());
+ }
+
+ if (expectedAfter != null) {
+ assertTrue(candidateNode.getDataAfter().isPresent());
+ assertEquals(expectedAfter, candidateNode.getDataAfter().get());
+ } else {
+ assertFalse(candidateNode.getDataAfter().isPresent());
+ }
+
+ assertEquals(expectedMod, candidateNode.getModificationType());
+ }
+
+ private DOMDataTreeChangeService getDOMDataTreeChangeService() {
+ final DOMDataBrokerExtension extension = domBroker.getSupportedExtensions()
+ .get(DOMDataTreeChangeService.class);
+ if (extension == null) {
+ return null;
+ }
+ DOMDataTreeChangeService dataTreeChangeService = null;
+ if (extension instanceof DOMDataTreeChangeService) {
+ dataTreeChangeService = (DOMDataTreeChangeService) extension;
+ }
+ return dataTreeChangeService;
+ }
+
+
+ static class CommitExecutorService extends ForwardingExecutorService {
+
+ ExecutorService delegate;
+
+ public CommitExecutorService(final ExecutorService delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ protected ExecutorService delegate() {
+ return delegate;
+ }
+ }
+
+ static class TestDataTreeListener implements DOMDataTreeChangeListener {
+
+ private List<Collection<DataTreeCandidate>> receivedChanges = new ArrayList<>();
+ private CountDownLatch latch;
+
+ public TestDataTreeListener(final CountDownLatch latch) {
+ this.latch = latch;
+ }
+
+ @Override
+ public void onDataTreeChanged(@Nonnull final Collection<DataTreeCandidate> changes) {
+ receivedChanges.add(changes);
+ latch.countDown();
+ }
+
+ public List<Collection<DataTreeCandidate>> getReceivedChanges() {
+ return receivedChanges;
+ }
+ }
+}
* Create rpc implementation capable of handling RPC for monitoring and notifications even before the schemas of remote device are downloaded
*/
static NetconfDeviceRpc getRpcForInitialization(final NetconfDeviceCommunicator listener) {
- return new NetconfDeviceRpc(INIT_SCHEMA_CTX, listener, new NetconfMessageTransformer(INIT_SCHEMA_CTX));
+ return new NetconfDeviceRpc(INIT_SCHEMA_CTX, listener, new NetconfMessageTransformer(INIT_SCHEMA_CTX, false));
}
@VisibleForTesting
void handleSalInitializationSuccess(final SchemaContext result, final NetconfSessionPreferences remoteSessionCapabilities, final DOMRpcService deviceRpc) {
- messageTransformer = new NetconfMessageTransformer(result);
+ messageTransformer = new NetconfMessageTransformer(result, true);
updateTransformer(messageTransformer);
// salFacade.onDeviceConnected has to be called before the notification handler is initialized
}
private NetconfDeviceRpc getDeviceSpecificRpc(final SchemaContext result) {
- return new NetconfDeviceRpc(result, listener, new NetconfMessageTransformer(result));
+ return new NetconfDeviceRpc(result, listener, new NetconfMessageTransformer(result, true));
}
private Collection<SourceIdentifier> stripMissingSource(final Collection<SourceIdentifier> requiredSources, final SourceIdentifier sIdToRemove) {
private final Multimap<QName, NotificationDefinition> mappedNotifications;
private final DomToNormalizedNodeParserFactory parserFactory;
- public NetconfMessageTransformer(final SchemaContext schemaContext) {
+ public NetconfMessageTransformer(final SchemaContext schemaContext, final boolean strictParsing) {
this.counter = new MessageCounter();
this.schemaContext = schemaContext;
- parserFactory = DomToNormalizedNodeParserFactory.getInstance(XmlUtils.DEFAULT_XML_CODEC_PROVIDER, schemaContext);
+ parserFactory = DomToNormalizedNodeParserFactory.getInstance(XmlUtils.DEFAULT_XML_CODEC_PROVIDER, schemaContext, strictParsing);
mappedRpcs = Maps.uniqueIndex(schemaContext.getOperations(), QNAME_FUNCTION);
mappedNotifications = Multimaps.index(schemaContext.getNotifications(), QNAME_NOREV_FUNCTION);
final DataSchemaNode schemasNode = ((ContainerSchemaNode) NetconfDevice.INIT_SCHEMA_CTX.getDataChildByName("netconf-state")).getDataChildByName("schemas");
final Document schemasXml = XmlUtil.readXmlToDocument(getClass().getResourceAsStream("/netconf-state.schemas.payload.xml"));
- final ToNormalizedNodeParser<Element, ContainerNode, ContainerSchemaNode> containerNodeParser = DomToNormalizedNodeParserFactory.getInstance(XmlUtils.DEFAULT_XML_CODEC_PROVIDER, NetconfDevice.INIT_SCHEMA_CTX).getContainerNodeParser();
+ final ToNormalizedNodeParser<Element, ContainerNode, ContainerSchemaNode> containerNodeParser = DomToNormalizedNodeParserFactory.getInstance(XmlUtils.DEFAULT_XML_CODEC_PROVIDER, NetconfDevice.INIT_SCHEMA_CTX, false).getContainerNodeParser();
final ContainerNode compositeNodeSchemas = containerNodeParser.parse(Collections.singleton(schemasXml.getDocumentElement()), (ContainerSchemaNode) schemasNode);
final NetconfStateSchemas schemas = NetconfStateSchemas.create(new RemoteDeviceId("device", new InetSocketAddress(99)), compositeNodeSchemas);
public void setup() throws Exception {
final SchemaContext schemaContext = getNotificationSchemaContext(getClass());
- messageTransformer = new NetconfMessageTransformer(schemaContext);
+ messageTransformer = new NetconfMessageTransformer(schemaContext, true);
final DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
factory.setNamespaceAware(true);
cfgCtx = parser.resolveSchemaContext(Sets.union(configModules, notifModules));
assertNotNull(cfgCtx);
- messageTransformer = new NetconfMessageTransformer(cfgCtx);
+ messageTransformer = new NetconfMessageTransformer(cfgCtx, true);
}
private LeafNode<Object> buildLeaf(final QName running, final Object value) {
}
private NetconfMessageTransformer getTransformer(final SchemaContext schema) {
- return new NetconfMessageTransformer(schema);
+ return new NetconfMessageTransformer(schema, true);
}
@Test
import static org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.OPERATIONAL;
import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.ArrayList;
import javax.ws.rs.core.Response.Status;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
-import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizationException;
-import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizationOperation;
-import org.opendaylight.controller.md.sal.common.impl.util.compat.DataNormalizer;
import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
import org.opendaylight.controller.md.sal.dom.api.DOMDataChangeListener;
import org.opendaylight.controller.md.sal.dom.api.DOMDataReadTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.opendaylight.yangtools.yang.model.api.SchemaPath;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// PUT configuration
public CheckedFuture<Void, TransactionCommitFailedException> commitConfigurationDataPut(
- final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload) {
+ final SchemaContext globalSchema, final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload) {
checkPreconditions();
- final DataNormalizationOperation<?> rootOp = ControllerContext.getInstance().getRootOperation();
- return putDataViaTransaction(domDataBroker.newReadWriteTransaction(), CONFIGURATION, path, payload, rootOp);
+ return putDataViaTransaction(domDataBroker.newReadWriteTransaction(), CONFIGURATION, path, payload, globalSchema);
}
public CheckedFuture<Void, TransactionCommitFailedException> commitConfigurationDataPut(
final DOMMountPoint mountPoint, final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload) {
final Optional<DOMDataBroker> domDataBrokerService = mountPoint.getService(DOMDataBroker.class);
if (domDataBrokerService.isPresent()) {
- final DataNormalizationOperation<?> rootOp = new DataNormalizer(mountPoint.getSchemaContext()).getRootOperation();
return putDataViaTransaction(domDataBrokerService.get().newReadWriteTransaction(), CONFIGURATION, path,
- payload, rootOp);
+ payload, mountPoint.getSchemaContext());
}
throw new RestconfDocumentedException("DOM data broker service isn't available for mount point.");
}
// POST configuration
public CheckedFuture<Void, TransactionCommitFailedException> commitConfigurationDataPost(
- final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload) {
+ final SchemaContext globalSchema, final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload) {
checkPreconditions();
- final DataNormalizationOperation<?> rootOp = ControllerContext.getInstance().getRootOperation();
- return postDataViaTransaction(domDataBroker.newReadWriteTransaction(), CONFIGURATION, path, payload, rootOp);
+ return postDataViaTransaction(domDataBroker.newReadWriteTransaction(), CONFIGURATION, path, payload, globalSchema);
}
public CheckedFuture<Void, TransactionCommitFailedException> commitConfigurationDataPost(
final DOMMountPoint mountPoint, final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload) {
final Optional<DOMDataBroker> domDataBrokerService = mountPoint.getService(DOMDataBroker.class);
if (domDataBrokerService.isPresent()) {
- final DataNormalizationOperation<?> rootOp = new DataNormalizer(mountPoint.getSchemaContext()).getRootOperation();
return postDataViaTransaction(domDataBrokerService.get().newReadWriteTransaction(), CONFIGURATION, path,
- payload, rootOp);
+ payload, mountPoint.getSchemaContext());
}
throw new RestconfDocumentedException("DOM data broker service isn't available for mount point.");
}
private CheckedFuture<Void, TransactionCommitFailedException> postDataViaTransaction(
final DOMDataReadWriteTransaction rWTransaction, final LogicalDatastoreType datastore,
- final YangInstanceIdentifier parentPath, final NormalizedNode<?, ?> payload, final DataNormalizationOperation<?> root) {
+ final YangInstanceIdentifier parentPath, final NormalizedNode<?, ?> payload, final SchemaContext schemaContext) {
// FIXME: This is doing correct post for container and list children
// not sure if this will work for choice case
final YangInstanceIdentifier path;
LOG.trace("It wasn't possible to get data loaded from datastore at path " + path);
}
- ensureParentsByMerge(datastore, path, rWTransaction, root);
+ ensureParentsByMerge(datastore, path, rWTransaction, schemaContext);
rWTransaction.merge(datastore, path, payload);
LOG.trace("Post " + datastore.name() + " via Restconf: {}", path);
return rWTransaction.submit();
private CheckedFuture<Void, TransactionCommitFailedException> putDataViaTransaction(
final DOMDataReadWriteTransaction writeTransaction, final LogicalDatastoreType datastore,
- final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload, final DataNormalizationOperation<?> root) {
+ final YangInstanceIdentifier path, final NormalizedNode<?, ?> payload, final SchemaContext schemaContext) {
LOG.trace("Put " + datastore.name() + " via Restconf: {}", path);
- ensureParentsByMerge(datastore, path, writeTransaction, root);
+ ensureParentsByMerge(datastore, path, writeTransaction, schemaContext);
writeTransaction.put(datastore, path, payload);
return writeTransaction.submit();
}
this.domDataBroker = domDataBroker;
}
- private final void ensureParentsByMerge(final LogicalDatastoreType store,
- final YangInstanceIdentifier normalizedPath, final DOMDataReadWriteTransaction rwTx,
- final DataNormalizationOperation<?> root) {
- final List<PathArgument> currentArguments = new ArrayList<>();
- final Iterator<PathArgument> iterator = normalizedPath.getPathArguments().iterator();
- DataNormalizationOperation<?> currentOp = root;
- while (iterator.hasNext()) {
- final PathArgument currentArg = iterator.next();
- try {
- currentOp = currentOp.getChild(currentArg);
- } catch (final DataNormalizationException e) {
- rwTx.cancel();
- throw new IllegalArgumentException(
- String.format("Invalid child encountered in path %s", normalizedPath), e);
- }
- currentArguments.add(currentArg);
- final YangInstanceIdentifier currentPath = YangInstanceIdentifier.create(currentArguments);
+ private void ensureParentsByMerge(final LogicalDatastoreType store,
+ final YangInstanceIdentifier normalizedPath, final DOMDataReadWriteTransaction rwTx, final SchemaContext schemaContext) {
+ final List<PathArgument> normalizedPathWithoutChildArgs = new ArrayList<>();
+ YangInstanceIdentifier rootNormalizedPath = null;
- final Boolean exists;
+ final Iterator<PathArgument> it = normalizedPath.getPathArguments().iterator();
- try {
-
- final CheckedFuture<Boolean, ReadFailedException> future = rwTx.exists(store, currentPath);
- exists = future.checkedGet();
- } catch (final ReadFailedException e) {
- LOG.error("Failed to read pre-existing data from store {} path {}", store, currentPath, e);
- rwTx.cancel();
- throw new IllegalStateException("Failed to read pre-existing data", e);
+ while(it.hasNext()) {
+ final PathArgument pathArgument = it.next();
+ if(rootNormalizedPath == null) {
+ rootNormalizedPath = YangInstanceIdentifier.create(pathArgument);
}
- if (!exists && iterator.hasNext()) {
- rwTx.merge(store, currentPath, currentOp.createDefault(currentArg));
+ // Skip last element, its not a parent
+ if(it.hasNext()) {
+ normalizedPathWithoutChildArgs.add(pathArgument);
}
}
+
+ // No parent structure involved, no need to ensure parents
+ if(normalizedPathWithoutChildArgs.isEmpty()) {
+ return;
+ }
+
+ Preconditions.checkArgument(rootNormalizedPath != null, "Empty path received");
+
+ final NormalizedNode<?, ?> parentStructure =
+ ImmutableNodes.fromInstanceId(schemaContext, YangInstanceIdentifier.create(normalizedPathWithoutChildArgs));
+ rwTx.merge(store, rootNormalizedPath, parentStructure);
}
}
if (mountPoint != null) {
broker.commitConfigurationDataPut(mountPoint, normalizedII, payload.getData()).checkedGet();
} else {
- broker.commitConfigurationDataPut(normalizedII, payload.getData()).checkedGet();
+ broker.commitConfigurationDataPut(controllerContext.getGlobalSchema(), normalizedII, payload.getData()).checkedGet();
}
break;
if (mountPoint != null) {
broker.commitConfigurationDataPost(mountPoint, normalizedII, payload.getData()).checkedGet();
} else {
- broker.commitConfigurationDataPost(normalizedII, payload.getData()).checkedGet();
+ broker.commitConfigurationDataPost(controllerContext.getGlobalSchema(), normalizedII, payload.getData()).checkedGet();
}
} catch(final RestconfDocumentedException e) {
throw e;
restconfImpl = RestconfImpl.getInstance();
restconfImpl.setBroker(brokerFacade);
restconfImpl.setControllerContext(controllerContext);
- when(brokerFacade.commitConfigurationDataPut(any(YangInstanceIdentifier.class), any(NormalizedNode.class)))
+ when(brokerFacade.commitConfigurationDataPut(any(SchemaContext.class), any(YangInstanceIdentifier.class), any(NormalizedNode.class)))
.thenReturn(mock(CheckedFuture.class));
}
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.opendaylight.yangtools.yang.model.api.SchemaPath;
/**
when(wTransaction.submit()).thenReturn(expFuture);
- final Future<Void> actualFuture = brokerFacade.commitConfigurationDataPut(instanceID, dummyNode);
+ final Future<Void> actualFuture = brokerFacade.commitConfigurationDataPut((SchemaContext)null, instanceID, dummyNode);
assertSame("commitConfigurationDataPut", expFuture, actualFuture);
when(rwTransaction.submit()).thenReturn(expFuture);
final CheckedFuture<Void, TransactionCommitFailedException> actualFuture = brokerFacade.commitConfigurationDataPost(
- YangInstanceIdentifier.builder().build(), dummyNode);
+ (SchemaContext)null, YangInstanceIdentifier.builder().build(), dummyNode);
assertSame("commitConfigurationDataPost", expFuture, actualFuture);
when(rwTransaction.read(eq(LogicalDatastoreType.CONFIGURATION), any(YangInstanceIdentifier.class))).thenReturn(
dummyNodeInFuture);
try {
- brokerFacade.commitConfigurationDataPost(instanceID, dummyNode);
+ // Schema context is only necessary for ensuring parent structure
+ brokerFacade.commitConfigurationDataPost((SchemaContext)null, instanceID, dummyNode);
} catch (final RestconfDocumentedException e) {
assertEquals("getErrorTag", RestconfError.ErrorTag.DATA_EXISTS, e.getErrors().get(0).getErrorTag());
throw e;
final RpcResult<TransactionStatus> rpcResult = new DummyRpcResult.Builder<TransactionStatus>().result(
TransactionStatus.COMMITED).build();
- when(brokerFacade.commitConfigurationDataPost(any(YangInstanceIdentifier.class), any(NormalizedNode.class)))
+ when(brokerFacade.commitConfigurationDataPost((SchemaContext)null, any(YangInstanceIdentifier.class), any(NormalizedNode.class)))
.thenReturn(mock(CheckedFuture.class));
final ArgumentCaptor<YangInstanceIdentifier> instanceIdCaptor = ArgumentCaptor.forClass(YangInstanceIdentifier.class);
// FIXME : NEVER test a nr. of call some service in complex test suite
// verify(brokerFacade, times(2))
verify(brokerFacade, times(1))
- .commitConfigurationDataPost(instanceIdCaptor.capture(), compNodeCaptor.capture());
+ .commitConfigurationDataPost((SchemaContext)null, instanceIdCaptor.capture(), compNodeCaptor.capture());
// identifier = "[(urn:ietf:params:xml:ns:yang:test-interface?revision=2014-07-01)interfaces, (urn:ietf:params:xml:ns:yang:test-interface?revision=2014-07-01)block]";
assertEquals(identifier, ImmutableList.copyOf(instanceIdCaptor.getValue().getPathArguments()).toString());
}
public void createConfigurationDataNullTest() throws UnsupportedEncodingException {
initMocking();
- when(brokerFacade.commitConfigurationDataPost(any(YangInstanceIdentifier.class),any(NormalizedNode.class)))
+ when(brokerFacade.commitConfigurationDataPost(any(SchemaContext.class), any(YangInstanceIdentifier.class),any(NormalizedNode.class)))
.thenReturn(Futures.<Void, TransactionCommitFailedException>immediateCheckedFuture(null));
//FIXME : find who is set schemaContext
doThrow(OptimisticLockFailedException.class).
when(brokerFacade).commitConfigurationDataPut(
- any(YangInstanceIdentifier.class), any(NormalizedNode.class));
+ any(SchemaContext.class), any(YangInstanceIdentifier.class), any(NormalizedNode.class));
assertEquals(500, put(uri, MediaType.APPLICATION_XML, xmlData));
doThrow(OptimisticLockFailedException.class).doReturn(mock(CheckedFuture.class)).
when(brokerFacade).commitConfigurationDataPut(
- any(YangInstanceIdentifier.class), any(NormalizedNode.class));
+ any(SchemaContext.class), any(YangInstanceIdentifier.class), any(NormalizedNode.class));
assertEquals(200, put(uri, MediaType.APPLICATION_XML, xmlData));
}
doThrow(TransactionCommitFailedException.class).
when(brokerFacade).commitConfigurationDataPut(
- any(YangInstanceIdentifier.class), any(NormalizedNode.class));
+ (SchemaContext)null, any(YangInstanceIdentifier.class), any(NormalizedNode.class));
assertEquals(500, put(uri, MediaType.APPLICATION_XML, xmlData));
}
private void mockCommitConfigurationDataPutMethod(final boolean noErrors) {
if (noErrors) {
doReturn(mock(CheckedFuture.class)).when(brokerFacade).commitConfigurationDataPut(
- any(YangInstanceIdentifier.class), any(NormalizedNode.class));
+ any(SchemaContext.class), any(YangInstanceIdentifier.class), any(NormalizedNode.class));
} else {
doThrow(RestconfDocumentedException.class).when(brokerFacade).commitConfigurationDataPut(
- any(YangInstanceIdentifier.class), any(NormalizedNode.class));
+ any(SchemaContext.class), any(YangInstanceIdentifier.class), any(NormalizedNode.class));
}
}
private static final Logger LOG = LoggerFactory.getLogger(SubtreeFilter.class);
static Document applySubtreeFilter(Document requestDocument, Document rpcReply) throws NetconfDocumentedException {
- // FIXME: rpcReply document must be reread otherwise some nodes do not inherit namespaces. (services/service)
- try {
- rpcReply = XmlUtil.readXmlToDocument(XmlUtil.toString(rpcReply, true));
- } catch (SAXException | IOException e) {
- LOG.error("Cannot transform document", e);
- throw new NetconfDocumentedException("Cannot transform document");
- }
-
OperationNameAndNamespace operationNameAndNamespace = new OperationNameAndNamespace(requestDocument);
if (XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0.equals(operationNameAndNamespace.getNamespace()) &&
XmlNetconfConstants.GET.equals(operationNameAndNamespace.getOperationName()) ||
// not implement filtering.
Optional<XmlElement> maybeFilter = operationNameAndNamespace.getOperationElement().getOnlyChildElementOptionally(
XmlNetconfConstants.FILTER, XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0);
- if (maybeFilter.isPresent() && (
- "subtree".equals(maybeFilter.get().getAttribute("type"))||
- "subtree".equals(maybeFilter.get().getAttribute("type", XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0)))
- ) {
+ if (!maybeFilter.isPresent()) {
+ return rpcReply;
+ }
+ // FIXME: rpcReply document must be reread otherwise some nodes do not inherit namespaces. (services/service)
+ try {
+ rpcReply = XmlUtil.readXmlToDocument(XmlUtil.toString(rpcReply, true));
+ } catch (SAXException | IOException e) {
+ LOG.error("Cannot transform document", e);
+ throw new NetconfDocumentedException("Cannot transform document" + e);
+ }
+ XmlElement filter = maybeFilter.get();
+ if ("subtree".equals(filter.getAttribute("type"))||
+ "subtree".equals(filter.getAttribute("type", XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_BASE_1_0))) {
// do
return filtered(maybeFilter.get(), rpcReply);
}
}
+
return rpcReply; // return identical document
}
Code Review / controller.git / commitdiff