X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-binding-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fbinding%2Fimpl%2FNotificationBrokerImpl.xtend;h=52aa8d029066b3093a90300f6bb078886b2b3d7b;hp=22db73526e2e2e386636cc8b45320b3c246f24ad;hb=ce2938e5616a170ca2fb2f5b478b3b2ceff832a1;hpb=738d46bed116293e3e42171ad2035ab805b0b2be diff --git a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.xtend b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.xtend index 22db73526e..52aa8d0290 100644 --- a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.xtend +++ b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.xtend @@ -1,52 +1,202 @@ -/* - * 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 +/* + * 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 com.google.common.collect.HashMultimap +import com.google.common.collect.Multimap +import java.util.Collection +import java.util.Collections +import java.util.concurrent.Callable +import java.util.concurrent.ExecutorService +import org.opendaylight.controller.sal.binding.api.NotificationListener +import org.opendaylight.controller.sal.binding.api.NotificationProviderService +import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker +import org.opendaylight.yangtools.concepts.AbstractObjectRegistration +import org.opendaylight.yangtools.concepts.ListenerRegistration +import org.opendaylight.yangtools.concepts.Registration +import org.opendaylight.yangtools.yang.binding.Notification +import org.slf4j.LoggerFactory +import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder import com.google.common.collect.Multimaps -import org.opendaylight.controller.sal.binding.api.NotificationProviderService -import org.opendaylight.yangtools.yang.binding.Notification -import com.google.common.collect.Multimap -import org.opendaylight.controller.sal.binding.api.NotificationListener -import com.google.common.collect.HashMultimap -import java.util.concurrent.ExecutorService -import java.util.Collection - -class NotificationBrokerImpl implements NotificationProviderService { - - val Multimap, NotificationListener> listeners; - val ExecutorService executor; - - new(ExecutorService executor) { - listeners = HashMultimap.create() - this.executor = executor; - } - - override addNotificationListener(Class notificationType, - NotificationListener listener) { - listeners.put(notificationType, listener) - } - - override removeNotificationListener(Class notificationType, - NotificationListener listener) { - listeners.remove(notificationType, listener) - } - - override notify(Notification notification) { - notification.notificationTypes.forEach [ - listeners.get(it as Class)?.notifyAll(notification) - ] - } - - def getNotificationTypes(Notification notification) { - notification.class.interfaces.filter[it != Notification && Notification.isAssignableFrom(it)] - } - - @SuppressWarnings("unchecked") - def notifyAll(Collection> listeners, Notification notification) { - listeners.forEach[(it as NotificationListener).onNotification(notification)] - } -} +class NotificationBrokerImpl implements NotificationProviderService, AutoCloseable { + + val Multimap, NotificationListener> listeners; + + @Property + var ExecutorService executor; + + new() { + listeners = Multimaps.synchronizedSetMultimap(HashMultimap.create()) + } + + @Deprecated + new(ExecutorService executor) { + listeners = Multimaps.synchronizedSetMultimap(HashMultimap.create()) + this.executor = executor; + } + + @Deprecated + override addNotificationListener(Class notificationType, + NotificationListener listener) { + listeners.put(notificationType, listener) + } + + @Deprecated + override removeNotificationListener(Class notificationType, + NotificationListener listener) { + listeners.remove(notificationType, listener) + } + + override notify(Notification notification) { + publish(notification) + } + + def getNotificationTypes(Notification notification) { + notification.class.interfaces.filter[it != Notification && Notification.isAssignableFrom(it)] + } + + @SuppressWarnings("unchecked") + private def notifyAll(Collection> listeners, Notification notification) { + listeners.forEach[(it as NotificationListener).onNotification(notification)] + } + + @Deprecated + override addNotificationListener(org.opendaylight.yangtools.yang.binding.NotificationListener listener) { + throw new UnsupportedOperationException("Deprecated method. Use registerNotificationListener instead."); + + } + + @Deprecated + override removeNotificationListener(org.opendaylight.yangtools.yang.binding.NotificationListener listener) { + throw new UnsupportedOperationException( + "Deprecated method. Use RegisterNotificationListener returned value to close registration.") + } + + @Deprecated + override notify(Notification notification, ExecutorService service) { + publish(notification, service) + } + + override publish(Notification notification) { + publish(notification, executor) + } + + override publish(Notification notification, ExecutorService service) { + val allTypes = notification.notificationTypes + + var Iterable> listenerToNotify = Collections.emptySet(); + for (type : allTypes) { + listenerToNotify = listenerToNotify + listeners.get(type as Class) + } + val tasks = listenerToNotify.map[new NotifyTask(it, notification)].toSet; + executor.invokeAll(tasks); + } + + override registerNotificationListener(Class notificationType, + NotificationListener listener) { + val reg = new GenericNotificationRegistration(notificationType, listener, this); + listeners.put(notificationType, listener); + return reg; + } + + override registerNotificationListener( + org.opendaylight.yangtools.yang.binding.NotificationListener listener) { + val invoker = SingletonHolder.INVOKER_FACTORY.invokerFor(listener); + for (notifyType : invoker.supportedNotifications) { + listeners.put(notifyType, invoker.invocationProxy) + } + val registration = new GeneratedListenerRegistration(listener, invoker,this); + return registration as Registration; + } + + protected def unregisterListener(GenericNotificationRegistration reg) { + listeners.remove(reg.type, reg.instance); + } + + protected def unregisterListener(GeneratedListenerRegistration reg) { + for (notifyType : reg.invoker.supportedNotifications) { + listeners.remove(notifyType, reg.invoker.invocationProxy) + } + } + + override close() { + //FIXME: implement properly. + } + +} + +class GenericNotificationRegistration extends AbstractObjectRegistration> implements ListenerRegistration> { + + @Property + val Class type; + + var NotificationBrokerImpl notificationBroker; + + public new(Class type, NotificationListener instance, NotificationBrokerImpl broker) { + super(instance); + _type = type; + notificationBroker = broker; + } + + override protected removeRegistration() { + notificationBroker.unregisterListener(this); + notificationBroker = null; + } +} + +class GeneratedListenerRegistration extends AbstractObjectRegistration implements ListenerRegistration { + + @Property + val NotificationInvoker invoker; + + var NotificationBrokerImpl notificationBroker; + + + new(org.opendaylight.yangtools.yang.binding.NotificationListener instance, NotificationInvoker invoker, NotificationBrokerImpl broker) { + super(instance); + _invoker = invoker; + notificationBroker = broker; + } + + override protected removeRegistration() { + notificationBroker.unregisterListener(this); + notificationBroker = null; + invoker.close(); + } +} + +@Data +class NotifyTask implements Callable { + + private static val log = LoggerFactory.getLogger(NotifyTask); + + @SuppressWarnings("rawtypes") + val NotificationListener listener; + val Notification notification; + + override call() { + //Only logging the complete notification in debug mode + try { + if(log.isDebugEnabled){ + log.debug("Delivering notification {} to {}",notification,listener); + } else { + log.trace("Delivering notification {} to {}",notification.class.name,listener); + } + listener.onNotification(notification); + if(log.isDebugEnabled){ + log.debug("Notification delivered {} to {}",notification,listener); + } else { + log.trace("Notification delivered {} to {}",notification.class.name,listener); + } + } catch (Exception e) { + log.error("Unhandled exception thrown by listener: {}", listener, e); + } + return null; + } + +}