d3b68002c3a64402a23d7fb20ae557165e4f9e72
[controller.git] / opendaylight / md-sal / sal-binding-broker / src / main / java / org / opendaylight / controller / sal / binding / impl / NotificationBrokerImpl.java
1 /**
2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8 package org.opendaylight.controller.sal.binding.impl;
9
10 import java.util.Arrays;
11 import java.util.Collections;
12 import java.util.HashSet;
13 import java.util.Set;
14 import java.util.concurrent.ExecutorService;
15
16 import org.opendaylight.controller.sal.binding.api.NotificationListener;
17 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
18 import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
19 import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker;
20 import org.opendaylight.yangtools.concepts.ListenerRegistration;
21 import org.opendaylight.yangtools.concepts.Registration;
22 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
23 import org.opendaylight.yangtools.yang.binding.Notification;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
26
27 import com.google.common.base.Preconditions;
28 import com.google.common.base.Predicate;
29 import com.google.common.collect.HashMultimap;
30 import com.google.common.collect.Iterables;
31 import com.google.common.collect.Multimap;
32 import com.google.common.collect.Multimaps;
33
34 public class NotificationBrokerImpl implements NotificationProviderService, AutoCloseable {
35     private static final Logger LOG = LoggerFactory.getLogger(NotificationBrokerImpl.class);
36
37     private final ListenerRegistry<NotificationInterestListener> interestListeners =
38             ListenerRegistry.create();
39
40     private final Multimap<Class<? extends Notification>, NotificationListener<?>> listeners =
41             Multimaps.synchronizedSetMultimap(HashMultimap.<Class<? extends Notification>, NotificationListener<?>>create());
42     private ExecutorService executor;
43
44     @Deprecated
45     public NotificationBrokerImpl(final ExecutorService executor) {
46         this.setExecutor(executor);
47     }
48
49     public void setExecutor(final ExecutorService executor) {
50         this.executor = Preconditions.checkNotNull(executor);
51     }
52
53     public Iterable<Class<?>> getNotificationTypes(final Notification notification) {
54         final Class<?>[] ifaces = notification.getClass().getInterfaces();
55         return Iterables.filter(Arrays.asList(ifaces), new Predicate<Class<?>>() {
56             @Override
57             public boolean apply(final Class<?> input) {
58                 if (Notification.class.equals(input)) {
59                     return false;
60                 }
61                 return Notification.class.isAssignableFrom(input);
62             }
63         });
64     }
65
66     @Override
67     public void publish(final Notification notification) {
68         this.publish(notification, executor);
69     }
70
71     @Override
72     public void publish(final Notification notification, final ExecutorService service) {
73         Iterable<NotificationListener<?>> listenerToNotify = Collections.emptySet();
74         for (final Class<?> type : getNotificationTypes(notification)) {
75             listenerToNotify = Iterables.concat(listenerToNotify, listeners.get(((Class<? extends Notification>) type)));
76         }
77
78         final Set<NotifyTask> tasks = new HashSet<>();
79         for (NotificationListener<?> l : listenerToNotify) {
80             tasks.add(new NotifyTask(l, notification));
81         }
82
83         for (final NotifyTask task : tasks) {
84             service.submit(task);
85         }
86     }
87
88     @Override
89     public <T extends Notification> Registration<NotificationListener<T>> registerNotificationListener(final Class<T> notificationType, final NotificationListener<T> listener) {
90         final GenericNotificationRegistration<T> reg = new GenericNotificationRegistration<T>(notificationType, listener, this);
91         this.listeners.put(notificationType, listener);
92         this.announceNotificationSubscription(notificationType);
93         return reg;
94     }
95
96     private void announceNotificationSubscription(final Class<? extends Notification> notification) {
97         for (final ListenerRegistration<NotificationInterestListener> listener : interestListeners) {
98             try {
99                 listener.getInstance().onNotificationSubscribtion(notification);
100             } catch (Exception e) {
101                 LOG.warn("Listener {} reported unexpected error on notification {}",
102                         listener.getInstance(), notification, e);
103             }
104         }
105     }
106
107     @Override
108     public Registration<org.opendaylight.yangtools.yang.binding.NotificationListener> registerNotificationListener(final org.opendaylight.yangtools.yang.binding.NotificationListener listener) {
109         final NotificationInvoker invoker = SingletonHolder.INVOKER_FACTORY.invokerFor(listener);
110         for (final Class<? extends Notification> notifyType : invoker.getSupportedNotifications()) {
111             listeners.put(notifyType, invoker.getInvocationProxy());
112             announceNotificationSubscription(notifyType);
113         }
114
115         return new GeneratedListenerRegistration(listener, invoker, this);
116     }
117
118     protected boolean unregisterListener(final GenericNotificationRegistration<?> reg) {
119         return listeners.remove(reg.getType(), reg.getInstance());
120     }
121
122     protected void unregisterListener(final GeneratedListenerRegistration reg) {
123         final NotificationInvoker invoker = reg.getInvoker();
124         for (final Class<? extends Notification> notifyType : invoker.getSupportedNotifications()) {
125             this.listeners.remove(notifyType, invoker.getInvocationProxy());
126         }
127     }
128
129     @Override
130     public void close() {
131     }
132
133     @Override
134     public ListenerRegistration<NotificationInterestListener> registerInterestListener(final NotificationInterestListener interestListener) {
135         final ListenerRegistration<NotificationInterestListener> registration = this.interestListeners.register(interestListener);
136         for (final Class<? extends Notification> notification : listeners.keySet()) {
137             interestListener.onNotificationSubscribtion(notification);
138         }
139         return registration;
140     }
141 }