opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/md/sal/binding/compat/HydrogenNotificationBrokerImpl.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.md.sal.binding.compat;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.collect.HashMultimap;
  12 import com.google.common.collect.Multimap;
  13 import java.util.Set;
  14 import java.util.concurrent.ExecutorService;
  15 import java.util.concurrent.atomic.AtomicReference;
  16 import javax.annotation.concurrent.GuardedBy;
  17 import org.opendaylight.controller.sal.binding.api.NotificationListener;
  18 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  19 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
  20 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  21 import org.opendaylight.yangtools.util.ListenerRegistry;
  22 import org.opendaylight.yangtools.yang.binding.Notification;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 @Deprecated
  27 public class HydrogenNotificationBrokerImpl implements NotificationProviderService, AutoCloseable {
  28     private static final Logger LOG = LoggerFactory.getLogger(HydrogenNotificationBrokerImpl.class);
  29
  30     private final ListenerRegistry<NotificationInterestListener> interestListeners =
  31             ListenerRegistry.create();
  32     private final AtomicReference<ListenerMapGeneration> listeners = new AtomicReference<>(new ListenerMapGeneration());
  33     private final ExecutorService executor;
  34
  35     public HydrogenNotificationBrokerImpl(final ExecutorService executor) {
  36         this.executor = Preconditions.checkNotNull(executor);
  37     }
  38
  39     @Override
  40     public void publish(final Notification notification) {
  41         publish(notification, executor);
  42     }
  43
  44     @Override
  45     public void publish(final Notification notification, final ExecutorService service) {
  46         for (final NotificationListenerRegistration<?> r : listeners.get().listenersFor(notification)) {
  47             service.submit(new NotifyTask(r, notification));
  48         }
  49     }
  50
  51     @GuardedBy("this")
  52     private Multimap<Class<? extends Notification>, NotificationListenerRegistration<?>> mutableListeners() {
  53         return HashMultimap.create(listeners.get().getListeners());
  54     }
  55
  56     private final void addRegistrations(final NotificationListenerRegistration<?>... registrations) {
  57         synchronized (this) {
  58             final Multimap<Class<? extends Notification>, NotificationListenerRegistration<?>> newListeners =
  59                     mutableListeners();
  60             for (final NotificationListenerRegistration<?> reg : registrations) {
  61                 newListeners.put(reg.getType(), reg);
  62             }
  63
  64             listeners.set(new ListenerMapGeneration(newListeners));
  65         }
  66
  67         // Notifications are dispatched out of lock...
  68         for (final NotificationListenerRegistration<?> reg : registrations) {
  69             announceNotificationSubscription(reg.getType());
  70         }
  71     }
  72
  73     private synchronized void removeRegistrations(final NotificationListenerRegistration<?>... registrations) {
  74         final Multimap<Class<? extends Notification>, NotificationListenerRegistration<?>> newListeners =
  75                 mutableListeners();
  76
  77         for (final NotificationListenerRegistration<?> reg : registrations) {
  78             newListeners.remove(reg.getType(), reg);
  79         }
  80
  81         listeners.set(new ListenerMapGeneration(newListeners));
  82     }
  83
  84     private void announceNotificationSubscription(final Class<? extends Notification> notification) {
  85         for (final ListenerRegistration<NotificationInterestListener> listener : interestListeners) {
  86             try {
  87                 listener.getInstance().onNotificationSubscribtion(notification);
  88             } catch (final Exception e) {
  89                 LOG.warn("Listener {} reported unexpected error on notification {}",
  90                         listener.getInstance(), notification, e);
  91             }
  92         }
  93     }
  94
  95     @Override
  96     public ListenerRegistration<NotificationInterestListener> registerInterestListener(final NotificationInterestListener interestListener) {
  97         final ListenerRegistration<NotificationInterestListener> registration = this.interestListeners.register(interestListener);
  98
  99         for (final Class<? extends Notification> notification : listeners.get().getKnownTypes()) {
 100             interestListener.onNotificationSubscribtion(notification);
 101         }
 102         return registration;
 103     }
 104
 105     @Override
 106     public <T extends Notification> NotificationListenerRegistration<T> registerNotificationListener(final Class<T> notificationType, final NotificationListener<T> listener) {
 107         final NotificationListenerRegistration<T> reg = new AbstractNotificationListenerRegistration<T>(notificationType, listener) {
 108             @Override
 109             protected void removeRegistration() {
 110                 removeRegistrations(this);
 111             }
 112         };
 113
 114         addRegistrations(reg);
 115         return reg;
 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 = NotificationInvoker.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 (final Class<? extends Notification> type : types) {
 127             regs[i] = new AggregatedNotificationListenerRegistration<Notification, Object>(type, invoker, 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 (final ListenerRegistration<?> reg : regs) {
 145                     reg.close();
 146                 }
 147             }
 148         };
 149     }
 150
 151     @Override
 152     public void close() {
 153     }
 154
 155 }