X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-binding-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fbinding%2Fimpl%2FNotificationBrokerImpl.xtend;h=cf339ee4f474e032ff8aca805f58eb5fe1f97d3b;hb=abf4ba1b0854fdd8b1a15c8c4481cf4b0bc93068;hp=22db73526e2e2e386636cc8b45320b3c246f24ad;hpb=fe024ad74b8656c3ee61b9ddff6009a779aa2189;p=controller.git

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..cf339ee4f4 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
@@ -14,31 +14,42 @@ import org.opendaylight.controller.sal.binding.api.NotificationListener
 import com.google.common.collect.HashMultimap
 import java.util.concurrent.ExecutorService
 import java.util.Collection
+import org.opendaylight.yangtools.concepts.Registration
+import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator
+import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory
+import org.opendaylight.yangtools.concepts.ListenerRegistration
+import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker
+import org.opendaylight.yangtools.concepts.AbstractObjectRegistration
+import java.util.Collections
+import org.slf4j.LoggerFactory
+import java.util.concurrent.Callable
 
-class NotificationBrokerImpl implements NotificationProviderService {
+class NotificationBrokerImpl implements NotificationProviderService, AutoCloseable {
 
     val Multimap<Class<? extends Notification>, NotificationListener<?>> listeners;
-    val ExecutorService executor;
+
+    @Property
+    var ExecutorService executor;
 
     new(ExecutorService executor) {
         listeners = HashMultimap.create()
         this.executor = executor;
     }
 
+    @Deprecated
     override <T extends Notification> addNotificationListener(Class<T> notificationType,
         NotificationListener<T> listener) {
         listeners.put(notificationType, listener)
     }
 
+    @Deprecated
     override <T extends Notification> removeNotificationListener(Class<T> notificationType,
         NotificationListener<T> listener) {
         listeners.remove(notificationType, listener)
     }
 
     override notify(Notification notification) {
-        notification.notificationTypes.forEach [
-            listeners.get(it as Class<? extends Notification>)?.notifyAll(notification)
-        ]
+        publish(notification)
     }
 
     def getNotificationTypes(Notification notification) {
@@ -46,7 +57,132 @@ class NotificationBrokerImpl implements NotificationProviderService {
     }
 
     @SuppressWarnings("unchecked")
-    def notifyAll(Collection<NotificationListener<?>> listeners, Notification notification) {
+    private def notifyAll(Collection<NotificationListener<?>> 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<NotificationListener<?>> listenerToNotify = Collections.emptySet();
+        for (type : allTypes) {
+            listenerToNotify = listenerToNotify + listeners.get(type as Class<? extends Notification>)
+        }
+        val tasks = listenerToNotify.map[new NotifyTask(it, notification)].toSet;
+        executor.invokeAll(tasks);
+    }
+
+    override <T extends Notification> registerNotificationListener(Class<T> notificationType,
+        NotificationListener<T> listener) {
+        val reg = new GenericNotificationRegistration<T>(notificationType, listener, this);
+        listeners.put(notificationType, listener);
+        return reg;
+    }
+
+    override registerNotificationListener(
+        org.opendaylight.yangtools.yang.binding.NotificationListener listener) {
+        val invoker = BindingAwareBrokerImpl.generator.invokerFactory.invokerFor(listener);
+        for (notifyType : invoker.supportedNotifications) {
+            listeners.put(notifyType, invoker.invocationProxy)
+        }
+        val registration = new GeneratedListenerRegistration(listener, invoker,this);
+        return registration as Registration<org.opendaylight.yangtools.yang.binding.NotificationListener>;
+    }
+
+    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<T extends Notification> extends AbstractObjectRegistration<NotificationListener<T>> implements ListenerRegistration<NotificationListener<T>> {
+
+    @Property
+    val Class<T> type;
+
+    var NotificationBrokerImpl notificationBroker;
+
+    public new(Class<T> type, NotificationListener<T> instance, NotificationBrokerImpl broker) {
+        super(instance);
+        _type = type;
+        notificationBroker = broker;
+    }
+
+    override protected removeRegistration() {
+        notificationBroker.unregisterListener(this);
+        notificationBroker = null;
+    }
+}
+
+class GeneratedListenerRegistration extends AbstractObjectRegistration<org.opendaylight.yangtools.yang.binding.NotificationListener> implements ListenerRegistration<org.opendaylight.yangtools.yang.binding.NotificationListener> {
+
+    @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<Object> {
+
+    private static val log = LoggerFactory.getLogger(NotifyTask);
+
+    val NotificationListener listener;
+    val Notification notification;
+
+    override call() {
+        try {
+            log.info("Delivering notification {} to {}",notification,listener);
+            listener.onNotification(notification);
+            log.info("Notification delivered {} to {}",notification,listener);
+        } catch (Exception e) {
+            log.error("Unhandled exception thrown by listener: {}", listener, e);
+        }
+        return null;
+    }
+
 }