From: Robert Varga Date: Mon, 2 Jun 2014 11:26:47 +0000 (+0200) Subject: BUG-1120: introduce NotificationListenerMap X-Git-Tag: release/helium~724 X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=commitdiff_plain;h=16dad33d563a80c98aa261bafc7cc3cf472fffb0;ds=sidebyside BUG-1120: introduce NotificationListenerMap Introduces encapsulation of the set of listeners, such that modifications and lookup are isolated fromt the rest of the implementation -- thus clearly defining the entry points. Also fixes a possible iteration failure if concurrent modification happens while a interest listener is being added. Change-Id: I96e0260d372425d261b352b229ae293e638c287c Signed-off-by: Robert Varga --- diff --git a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.java b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.java index 716670f5a3..49c24f9dc5 100644 --- a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.java +++ b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationBrokerImpl.java @@ -7,9 +7,6 @@ */ package org.opendaylight.controller.sal.binding.impl; -import java.util.Arrays; -import java.util.Collection; -import java.util.HashSet; import java.util.Set; import java.util.concurrent.ExecutorService; @@ -25,70 +22,56 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.common.base.Preconditions; -import com.google.common.base.Predicate; -import com.google.common.collect.HashMultimap; -import com.google.common.collect.Iterables; -import com.google.common.collect.Multimap; -import com.google.common.collect.Multimaps; public class NotificationBrokerImpl implements NotificationProviderService, AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(NotificationBrokerImpl.class); private final ListenerRegistry interestListeners = ListenerRegistry.create(); - private final Multimap, NotificationListenerRegistration> listeners = - Multimaps.synchronizedSetMultimap(HashMultimap., NotificationListenerRegistration>create()); + private final NotificationListenerMap listeners = new NotificationListenerMap(); private final ExecutorService executor; public NotificationBrokerImpl(final ExecutorService executor) { this.executor = Preconditions.checkNotNull(executor); } - public Iterable> getNotificationTypes(final Notification notification) { - final Class[] ifaces = notification.getClass().getInterfaces(); - return Iterables.filter(Arrays.asList(ifaces), new Predicate>() { - @Override - public boolean apply(final Class input) { - if (Notification.class.equals(input)) { - return false; - } - return Notification.class.isAssignableFrom(input); - } - }); - } - @Override public void publish(final Notification notification) { - this.publish(notification, executor); + publish(notification, executor); } @Override public void publish(final Notification notification, final ExecutorService service) { - final Set> toNotify = new HashSet<>(); - - for (final Class type : getNotificationTypes(notification)) { - final Collection> l = listeners.get((Class) type); - if (l != null) { - toNotify.addAll(l); - } - } - - for (NotificationListenerRegistration r : toNotify) { + for (NotificationListenerRegistration r : listeners.listenersFor(notification)) { service.submit(new NotifyTask(r, notification)); } } - private void addRegistrations(final NotificationListenerRegistration... registrations) { + private final void addRegistrations(final NotificationListenerRegistration... registrations) { + listeners.addRegistrations(registrations); for (NotificationListenerRegistration reg : registrations) { - listeners.put(reg.getType(), reg); - this.announceNotificationSubscription(reg.getType()); + announceNotificationSubscription(reg.getType()); } } - void removeRegistrations(final NotificationListenerRegistration... registrations) { - for (NotificationListenerRegistration reg : registrations) { - listeners.remove(reg.getType(), reg); + private void announceNotificationSubscription(final Class notification) { + for (final ListenerRegistration listener : interestListeners) { + try { + listener.getInstance().onNotificationSubscribtion(notification); + } catch (Exception e) { + LOG.warn("Listener {} reported unexpected error on notification {}", + listener.getInstance(), notification, e); + } + } + } + + @Override + public ListenerRegistration registerInterestListener(final NotificationInterestListener interestListener) { + final ListenerRegistration registration = this.interestListeners.register(interestListener); + for (final Class notification : listeners.getKnownTypes()) { + interestListener.onNotificationSubscribtion(notification); } + return registration; } @Override @@ -96,7 +79,7 @@ public class NotificationBrokerImpl implements NotificationProviderService, Auto final NotificationListenerRegistration reg = new AbstractNotificationListenerRegistration(notificationType, listener) { @Override protected void removeRegistration() { - removeRegistrations(this); + listeners.removeRegistrations(this); } }; @@ -104,17 +87,6 @@ public class NotificationBrokerImpl implements NotificationProviderService, Auto return reg; } - private void announceNotificationSubscription(final Class notification) { - for (final ListenerRegistration listener : interestListeners) { - try { - listener.getInstance().onNotificationSubscribtion(notification); - } catch (Exception e) { - LOG.warn("Listener {} reported unexpected error on notification {}", - listener.getInstance(), notification, e); - } - } - } - @Override public ListenerRegistration registerNotificationListener(final org.opendaylight.yangtools.yang.binding.NotificationListener listener) { final NotificationInvoker invoker = SingletonHolder.INVOKER_FACTORY.invokerFor(listener); @@ -140,7 +112,7 @@ public class NotificationBrokerImpl implements NotificationProviderService, Auto return new AbstractListenerRegistration(listener) { @Override protected void removeRegistration() { - removeRegistrations(regs); + listeners.removeRegistrations(regs); for (ListenerRegistration reg : regs) { reg.close(); } @@ -152,12 +124,4 @@ public class NotificationBrokerImpl implements NotificationProviderService, Auto public void close() { } - @Override - public ListenerRegistration registerInterestListener(final NotificationInterestListener interestListener) { - final ListenerRegistration registration = this.interestListeners.register(interestListener); - for (final Class notification : listeners.keySet()) { - interestListener.onNotificationSubscribtion(notification); - } - return registration; - } } diff --git a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationListenerMap.java b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationListenerMap.java new file mode 100644 index 0000000000..d3039fede8 --- /dev/null +++ b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/NotificationListenerMap.java @@ -0,0 +1,70 @@ +/** + * 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 java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +import org.opendaylight.yangtools.yang.binding.Notification; + +import com.google.common.base.Predicate; +import com.google.common.collect.HashMultimap; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; +import com.google.common.collect.Multimap; +import com.google.common.collect.Multimaps; + +final class NotificationListenerMap { + private final Multimap, NotificationListenerRegistration> listeners = + Multimaps.synchronizedSetMultimap(HashMultimap., NotificationListenerRegistration>create()); + + private static Iterable> getNotificationTypes(final Notification notification) { + final Class[] ifaces = notification.getClass().getInterfaces(); + return Iterables.filter(Arrays.asList(ifaces), new Predicate>() { + @Override + public boolean apply(final Class input) { + if (Notification.class.equals(input)) { + return false; + } + return Notification.class.isAssignableFrom(input); + } + }); + } + + Iterable> listenersFor(final Notification notification) { + final Set> toNotify = new HashSet<>(); + + for (final Class type : getNotificationTypes(notification)) { + final Collection> l = listeners.get((Class) type); + if (l != null) { + toNotify.addAll(l); + } + } + + return toNotify; + } + + Iterable> getKnownTypes() { + return ImmutableList.copyOf(listeners.keySet()); + } + + void addRegistrations(final NotificationListenerRegistration... registrations) { + for (NotificationListenerRegistration reg : registrations) { + listeners.put(reg.getType(), reg); + } + } + + void removeRegistrations(final NotificationListenerRegistration... registrations) { + for (NotificationListenerRegistration reg : registrations) { + listeners.remove(reg.getType(), reg); + } + } + +}