BUG-868: disallow instantiation without executor
[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.Collection;
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.AbstractListenerRegistration;
21 import org.opendaylight.yangtools.concepts.ListenerRegistration;
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     private final Multimap<Class<? extends Notification>, NotificationListenerRegistration<?>> listeners =
40             Multimaps.synchronizedSetMultimap(HashMultimap.<Class<? extends Notification>, NotificationListenerRegistration<?>>create());
41     private final ExecutorService executor;
42
43     public NotificationBrokerImpl(final ExecutorService executor) {
44         this.executor = Preconditions.checkNotNull(executor);
45     }
46
47     public Iterable<Class<?>> getNotificationTypes(final Notification notification) {
48         final Class<?>[] ifaces = notification.getClass().getInterfaces();
49         return Iterables.filter(Arrays.asList(ifaces), new Predicate<Class<?>>() {
50             @Override
51             public boolean apply(final Class<?> input) {
52                 if (Notification.class.equals(input)) {
53                     return false;
54                 }
55                 return Notification.class.isAssignableFrom(input);
56             }
57         });
58     }
59
60     @Override
61     public void publish(final Notification notification) {
62         this.publish(notification, executor);
63     }
64
65     @Override
66     public void publish(final Notification notification, final ExecutorService service) {
67         final Set<NotificationListenerRegistration<?>> toNotify = new HashSet<>();
68
69         for (final Class<?> type : getNotificationTypes(notification)) {
70             final Collection<NotificationListenerRegistration<?>> l = listeners.get((Class<? extends Notification>) type);
71             if (l != null) {
72                 toNotify.addAll(l);
73             }
74         }
75
76         for (NotificationListenerRegistration<?> r : toNotify) {
77             service.submit(new NotifyTask(r, notification));
78         }
79     }
80
81     private void addRegistrations(final NotificationListenerRegistration<?>... registrations) {
82         for (NotificationListenerRegistration<?> reg : registrations) {
83             listeners.put(reg.getType(), reg);
84             this.announceNotificationSubscription(reg.getType());
85         }
86     }
87
88     void removeRegistrations(final NotificationListenerRegistration<?>... registrations) {
89         for (NotificationListenerRegistration<?> reg : registrations) {
90             listeners.remove(reg.getType(), reg);
91         }
92     }
93
94     @Override
95     public <T extends Notification> NotificationListenerRegistration<T> registerNotificationListener(final Class<T> notificationType, final NotificationListener<T> listener) {
96         final NotificationListenerRegistration<T> reg = new AbstractNotificationListenerRegistration<T>(notificationType, listener) {
97             @Override
98             protected void removeRegistration() {
99                 removeRegistrations(this);
100             }
101         };
102
103         addRegistrations(reg);
104         return reg;
105     }
106
107     private void announceNotificationSubscription(final Class<? extends Notification> notification) {
108         for (final ListenerRegistration<NotificationInterestListener> listener : interestListeners) {
109             try {
110                 listener.getInstance().onNotificationSubscribtion(notification);
111             } catch (Exception e) {
112                 LOG.warn("Listener {} reported unexpected error on notification {}",
113                         listener.getInstance(), notification, e);
114             }
115         }
116     }
117
118     @Override
119     public ListenerRegistration<org.opendaylight.yangtools.yang.binding.NotificationListener> registerNotificationListener(final org.opendaylight.yangtools.yang.binding.NotificationListener listener) {
120         final NotificationInvoker invoker = SingletonHolder.INVOKER_FACTORY.invokerFor(listener);
121         final Set<Class<? extends Notification>> types = invoker.getSupportedNotifications();
122         final NotificationListenerRegistration<?>[] regs = new NotificationListenerRegistration<?>[types.size()];
123
124         // Populate the registrations...
125         int i = 0;
126         for (Class<? extends Notification> type : types) {
127             regs[i] = new AggregatedNotificationListenerRegistration<Notification, Object>(type, invoker.getInvocationProxy(), regs) {
128                 @Override
129                 protected void removeRegistration() {
130                     // Nothing to do, will be cleaned up by parent (below)
131                 }
132             };
133             ++i;
134         }
135
136         // ... now put them to use ...
137         addRegistrations(regs);
138
139         // ... finally return the parent registration
140         return new AbstractListenerRegistration<org.opendaylight.yangtools.yang.binding.NotificationListener>(listener) {
141             @Override
142             protected void removeRegistration() {
143                 removeRegistrations(regs);
144                 for (ListenerRegistration<?> reg : regs) {
145                     reg.close();
146                 }
147             }
148         };
149     }
150
151     @Override
152     public void close() {
153     }
154
155     @Override
156     public ListenerRegistration<NotificationInterestListener> registerInterestListener(final NotificationInterestListener interestListener) {
157         final ListenerRegistration<NotificationInterestListener> registration = this.interestListeners.register(interestListener);
158         for (final Class<? extends Notification> notification : listeners.keySet()) {
159             interestListener.onNotificationSubscribtion(notification);
160         }
161         return registration;
162     }
163 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.