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.Collection;
  11 import java.util.Collections;
  12 import java.util.Set;
  13 import java.util.concurrent.ExecutorService;
  14 import java.util.concurrent.Future;
  15
  16 import org.eclipse.xtext.xbase.lib.Conversions;
  17 import org.eclipse.xtext.xbase.lib.Exceptions;
  18 import org.eclipse.xtext.xbase.lib.Functions.Function0;
  19 import org.eclipse.xtext.xbase.lib.Functions.Function1;
  20 import org.eclipse.xtext.xbase.lib.IterableExtensions;
  21 import org.opendaylight.controller.sal.binding.api.NotificationListener;
  22 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  23 import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
  24 import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory.NotificationInvoker;
  25 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  26 import org.opendaylight.yangtools.concepts.Registration;
  27 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
  28 import org.opendaylight.yangtools.yang.binding.Notification;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 import com.google.common.base.Objects;
  33 import com.google.common.collect.HashMultimap;
  34 import com.google.common.collect.ImmutableSet;
  35 import com.google.common.collect.ImmutableSet.Builder;
  36 import com.google.common.collect.Iterables;
  37 import com.google.common.collect.Multimap;
  38 import com.google.common.collect.Multimaps;
  39 import com.google.common.collect.SetMultimap;
  40
  41 public class NotificationBrokerImpl implements NotificationProviderService, AutoCloseable {
  42     private final ListenerRegistry<NotificationInterestListener> interestListeners = new Function0<ListenerRegistry<NotificationInterestListener>>() {
  43         @Override
  44         public ListenerRegistry<NotificationInterestListener> apply() {
  45             ListenerRegistry<NotificationInterestListener> _create = ListenerRegistry.<NotificationInterestListener>create();
  46             return _create;
  47         }
  48     }.apply();
  49
  50     private final Multimap<Class<? extends Notification>,NotificationListener<? extends Object>> listeners;
  51
  52     private ExecutorService _executor;
  53
  54     public ExecutorService getExecutor() {
  55         return this._executor;
  56     }
  57
  58     public void setExecutor(final ExecutorService executor) {
  59         this._executor = executor;
  60     }
  61
  62     private final Logger logger = new Function0<Logger>() {
  63         @Override
  64         public Logger apply() {
  65             Logger _logger = LoggerFactory.getLogger(NotificationBrokerImpl.class);
  66             return _logger;
  67         }
  68     }.apply();
  69
  70     public NotificationBrokerImpl() {
  71         HashMultimap<Class<? extends Notification>,NotificationListener<? extends Object>> _create = HashMultimap.<Class<? extends Notification>, NotificationListener<? extends Object>>create();
  72         SetMultimap<Class<? extends Notification>,NotificationListener<? extends Object>> _synchronizedSetMultimap = Multimaps.<Class<? extends Notification>, NotificationListener<? extends Object>>synchronizedSetMultimap(_create);
  73         this.listeners = _synchronizedSetMultimap;
  74     }
  75
  76     @Deprecated
  77     public NotificationBrokerImpl(final ExecutorService executor) {
  78         HashMultimap<Class<? extends Notification>,NotificationListener<? extends Object>> _create = HashMultimap.<Class<? extends Notification>, NotificationListener<? extends Object>>create();
  79         SetMultimap<Class<? extends Notification>,NotificationListener<? extends Object>> _synchronizedSetMultimap = Multimaps.<Class<? extends Notification>, NotificationListener<? extends Object>>synchronizedSetMultimap(_create);
  80         this.listeners = _synchronizedSetMultimap;
  81         this.setExecutor(executor);
  82     }
  83
  84     public Iterable<Class<? extends Object>> getNotificationTypes(final Notification notification) {
  85         Class<? extends Notification> _class = notification.getClass();
  86         Class<? extends Object>[] _interfaces = _class.getInterfaces();
  87         final Function1<Class<? extends Object>,Boolean> _function = new Function1<Class<? extends Object>,Boolean>() {
  88             @Override
  89             public Boolean apply(final Class<? extends Object> it) {
  90                 boolean _and = false;
  91                 boolean _notEquals = (!Objects.equal(it, Notification.class));
  92                 if (!_notEquals) {
  93                     _and = false;
  94                 } else {
  95                     boolean _isAssignableFrom = Notification.class.isAssignableFrom(it);
  96                     _and = (_notEquals && _isAssignableFrom);
  97                 }
  98                 return Boolean.valueOf(_and);
  99             }
 100         };
 101         Iterable<Class<? extends Object>> _filter = IterableExtensions.<Class<? extends Object>>filter(((Iterable<Class<? extends Object>>)Conversions.doWrapArray(_interfaces)), _function);
 102         return _filter;
 103     }
 104
 105     @Override
 106     public void publish(final Notification notification) {
 107         ExecutorService _executor = this.getExecutor();
 108         this.publish(notification, _executor);
 109     }
 110
 111     @Override
 112     public void publish(final Notification notification, final ExecutorService service) {
 113         final Iterable<Class<? extends Object>> allTypes = this.getNotificationTypes(notification);
 114         Iterable<NotificationListener<? extends Object>> listenerToNotify = Collections.<NotificationListener<? extends Object>>emptySet();
 115         for (final Class<? extends Object> type : allTypes) {
 116             Collection<NotificationListener<? extends Object>> _get = this.listeners.get(((Class<? extends Notification>) type));
 117             Iterable<NotificationListener<? extends Object>> _plus = Iterables.<NotificationListener<? extends Object>>concat(listenerToNotify, _get);
 118             listenerToNotify = _plus;
 119         }
 120         final Function1<NotificationListener<? extends Object>,NotifyTask> _function = new Function1<NotificationListener<? extends Object>,NotifyTask>() {
 121             @Override
 122             public NotifyTask apply(final NotificationListener<? extends Object> it) {
 123                 NotifyTask _notifyTask = new NotifyTask(it, notification);
 124                 return _notifyTask;
 125             }
 126         };
 127         Iterable<NotifyTask> _map = IterableExtensions.<NotificationListener<? extends Object>, NotifyTask>map(listenerToNotify, _function);
 128         final Set<NotifyTask> tasks = IterableExtensions.<NotifyTask>toSet(_map);
 129         ExecutorService _executor = this.getExecutor();
 130         this.submitAll(_executor, tasks);
 131     }
 132
 133     public ImmutableSet<Future<Object>> submitAll(final ExecutorService service, final Set<NotifyTask> tasks) {
 134         final Builder<Future<Object>> ret = ImmutableSet.<Future<Object>>builder();
 135         for (final NotifyTask task : tasks) {
 136             Future<Object> _submit = service.<Object>submit(task);
 137             ret.add(_submit);
 138         }
 139         return ret.build();
 140     }
 141
 142     @Override
 143     public <T extends Notification> Registration<NotificationListener<T>> registerNotificationListener(final Class<T> notificationType, final NotificationListener<T> listener) {
 144         GenericNotificationRegistration<T> _genericNotificationRegistration = new GenericNotificationRegistration<T>(notificationType, listener, this);
 145         final GenericNotificationRegistration<T> reg = _genericNotificationRegistration;
 146         this.listeners.put(notificationType, listener);
 147         this.announceNotificationSubscription(notificationType);
 148         return reg;
 149     }
 150
 151     public void announceNotificationSubscription(final Class<? extends Notification> notification) {
 152         for (final ListenerRegistration<NotificationInterestListener> listener : this.interestListeners) {
 153             try {
 154                 NotificationInterestListener _instance = listener.getInstance();
 155                 _instance.onNotificationSubscribtion(notification);
 156             } catch (final Throwable _t) {
 157                 if (_t instanceof Exception) {
 158                     final Exception e = (Exception)_t;
 159                     String _message = e.getMessage();
 160                     this.logger.error("", _message);
 161                 } else {
 162                     throw Exceptions.sneakyThrow(_t);
 163                 }
 164             }
 165         }
 166     }
 167
 168     @Override
 169     public Registration<org.opendaylight.yangtools.yang.binding.NotificationListener> registerNotificationListener(final org.opendaylight.yangtools.yang.binding.NotificationListener listener) {
 170         final NotificationInvoker invoker = SingletonHolder.INVOKER_FACTORY.invokerFor(listener);
 171         Set<Class<? extends Notification>> _supportedNotifications = invoker.getSupportedNotifications();
 172         for (final Class<? extends Notification> notifyType : _supportedNotifications) {
 173             {
 174                 NotificationListener<Notification> _invocationProxy = invoker.getInvocationProxy();
 175                 this.listeners.put(notifyType, _invocationProxy);
 176                 this.announceNotificationSubscription(notifyType);
 177             }
 178         }
 179         GeneratedListenerRegistration _generatedListenerRegistration = new GeneratedListenerRegistration(listener, invoker, this);
 180         final GeneratedListenerRegistration registration = _generatedListenerRegistration;
 181         return (registration);
 182     }
 183
 184     protected boolean unregisterListener(final GenericNotificationRegistration<? extends Object> reg) {
 185         Class<? extends Notification> _type = reg.getType();
 186         NotificationListener<? extends Notification> _instance = reg.getInstance();
 187         boolean _remove = this.listeners.remove(_type, _instance);
 188         return _remove;
 189     }
 190
 191     protected void unregisterListener(final GeneratedListenerRegistration reg) {
 192         NotificationInvoker _invoker = reg.getInvoker();
 193         Set<Class<? extends Notification>> _supportedNotifications = _invoker.getSupportedNotifications();
 194         for (final Class<? extends Notification> notifyType : _supportedNotifications) {
 195             NotificationInvoker _invoker_1 = reg.getInvoker();
 196             NotificationListener<Notification> _invocationProxy = _invoker_1.getInvocationProxy();
 197             this.listeners.remove(notifyType, _invocationProxy);
 198         }
 199     }
 200
 201     @Override
 202     public void close() {
 203     }
 204
 205     @Override
 206     public ListenerRegistration<NotificationInterestListener> registerInterestListener(final NotificationInterestListener interestListener) {
 207         final ListenerRegistration<NotificationInterestListener> registration = this.interestListeners.register(interestListener);
 208         Set<Class<? extends Notification>> _keySet = this.listeners.keySet();
 209         for (final Class<? extends Notification> notification : _keySet) {
 210             interestListener.onNotificationSubscribtion(notification);
 211         }
 212         return registration;
 213     }
 214 }