opendaylight/sal/yang-prototype/sal/sal-broker-impl/src/main/java/org/opendaylight/controller/sal/core/impl/NotificationModule.java

   1 /*\r
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.\r
   3  *\r
   4  * This program and the accompanying materials are made available under the\r
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,\r
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html\r
   7  */\r
   8 package org.opendaylight.controller.sal.core.impl.notify;\r
   9 \r
  10 import java.util.Collection;\r
  11 import java.util.Collections;\r
  12 import java.util.HashSet;\r
  13 import java.util.Map;\r
  14 import java.util.Map.Entry;\r
  15 import java.util.Set;\r
  16 \r
  17 import org.opendaylight.controller.sal.core.api.BrokerService;\r
  18 import org.opendaylight.controller.sal.core.api.Broker.ConsumerSession;\r
  19 import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;\r
  20 import org.opendaylight.controller.sal.core.api.Consumer.ConsumerFunctionality;\r
  21 import org.opendaylight.controller.sal.core.api.Provider.ProviderFunctionality;\r
  22 import org.opendaylight.controller.sal.core.api.notify.NotificationListener;\r
  23 import org.opendaylight.controller.sal.core.api.notify.NotificationProviderService;\r
  24 import org.opendaylight.controller.sal.core.api.notify.NotificationService;\r
  25 import org.opendaylight.controller.sal.core.spi.BrokerModule;\r
  26 import org.opendaylight.controller.yang.common.QName;\r
  27 import org.opendaylight.controller.yang.data.api.CompositeNode;\r
  28 import org.slf4j.Logger;\r
  29 import org.slf4j.LoggerFactory;\r
  30 \r
  31 import com.google.common.collect.HashMultimap;\r
  32 import com.google.common.collect.ImmutableSet;\r
  33 import com.google.common.collect.Multimap;\r
  34 \r
  35 public class NotificationModule implements BrokerModule {\r
  36     private static Logger log = LoggerFactory\r
  37             .getLogger(NotificationModule.class);\r
  38 \r
  39     private Multimap<QName, NotificationListener> listeners = HashMultimap\r
  40             .create();\r
  41 \r
  42     private static final Set<Class<? extends BrokerService>> PROVIDED_SERVICE_TYPE = ImmutableSet\r
  43             .of((Class<? extends BrokerService>) NotificationService.class,\r
  44                     NotificationProviderService.class);\r
  45 \r
  46     private static final Set<Class<? extends ConsumerFunctionality>> SUPPORTED_CONSUMER_FUNCTIONALITY = ImmutableSet\r
  47             .of((Class<? extends ConsumerFunctionality>) NotificationListener.class,\r
  48                     NotificationListener.class); // Workaround: if we use the\r
  49                                                  // version of method with only\r
  50                                                  // one argument, the generics\r
  51                                                  // inference will not work\r
  52 \r
  53     @Override\r
  54     public Set<Class<? extends BrokerService>> getProvidedServices() {\r
  55         return PROVIDED_SERVICE_TYPE;\r
  56     }\r
  57 \r
  58     @Override\r
  59     public Set<Class<? extends ConsumerFunctionality>> getSupportedConsumerFunctionality() {\r
  60         return SUPPORTED_CONSUMER_FUNCTIONALITY;\r
  61     }\r
  62 \r
  63     @Override\r
  64     public <T extends BrokerService> T getServiceForSession(Class<T> service,\r
  65             ConsumerSession session) {\r
  66         if (NotificationProviderService.class.equals(service)\r
  67                 && session instanceof ProviderSession) {\r
  68             @SuppressWarnings("unchecked")\r
  69             T ret = (T) newNotificationProviderService(session);\r
  70             return ret;\r
  71         } else if (NotificationService.class.equals(service)) {\r
  72 \r
  73             @SuppressWarnings("unchecked")\r
  74             T ret = (T) newNotificationConsumerService(session);\r
  75             return ret;\r
  76         }\r
  77 \r
  78         throw new IllegalArgumentException(\r
  79                 "The requested session-specific service is not provided by this module.");\r
  80     }\r
  81 \r
  82     private void sendNotification(CompositeNode notification) {\r
  83         QName type = notification.getNodeType();\r
  84         Collection<NotificationListener> toNotify = listeners.get(type);\r
  85         log.info("Publishing notification " + type);\r
  86 \r
  87         if (toNotify == null) {\r
  88             // No listeners were registered - returns.\r
  89             return;\r
  90         }\r
  91 \r
  92         for (NotificationListener listener : toNotify) {\r
  93             try {\r
  94                 // FIXME: ensure that notification is immutable\r
  95                 listener.onNotification(notification);\r
  96             } catch (Exception e) {\r
  97                 log.error("Uncaught exception in NotificationListener", e);\r
  98             }\r
  99         }\r
 100 \r
 101     }\r
 102 \r
 103     private NotificationService newNotificationConsumerService(\r
 104             ConsumerSession session) {\r
 105         return new NotificationConsumerSessionImpl();\r
 106     }\r
 107 \r
 108     private NotificationProviderService newNotificationProviderService(\r
 109             ConsumerSession session) {\r
 110         return new NotificationProviderSessionImpl();\r
 111     }\r
 112 \r
 113     private class NotificationConsumerSessionImpl implements\r
 114             NotificationService {\r
 115 \r
 116         private Multimap<QName, NotificationListener> consumerListeners = HashMultimap\r
 117                 .create();\r
 118         private boolean closed = false;\r
 119 \r
 120         @Override\r
 121         public void addNotificationListener(QName notification,\r
 122                 NotificationListener listener) {\r
 123             checkSessionState();\r
 124             if (notification == null) {\r
 125                 throw new IllegalArgumentException(\r
 126                         "Notification type must not be null.");\r
 127             }\r
 128             if (listener == null) {\r
 129                 throw new IllegalArgumentException("Listener must not be null.");\r
 130             }\r
 131 \r
 132             consumerListeners.put(notification, listener);\r
 133             listeners.put(notification, listener);\r
 134             log.info("Registered listener for notification: " + notification);\r
 135         }\r
 136 \r
 137         @Override\r
 138         public void removeNotificationListener(QName notification,\r
 139                 NotificationListener listener) {\r
 140             checkSessionState();\r
 141             if (notification == null) {\r
 142                 throw new IllegalArgumentException(\r
 143                         "Notification type must not be null.");\r
 144             }\r
 145             if (listener == null) {\r
 146                 throw new IllegalArgumentException("Listener must not be null.");\r
 147             }\r
 148             consumerListeners.remove(notification, listener);\r
 149             listeners.remove(notification, listener);\r
 150         }\r
 151 \r
 152         @Override\r
 153         public void closeSession() {\r
 154             closed = true;\r
 155             Map<QName, Collection<NotificationListener>> toRemove = consumerListeners\r
 156                     .asMap();\r
 157             for (Entry<QName, Collection<NotificationListener>> entry : toRemove\r
 158                     .entrySet()) {\r
 159                 listeners.remove(entry.getKey(), entry.getValue());\r
 160             }\r
 161         }\r
 162 \r
 163         protected void checkSessionState() {\r
 164             if (closed)\r
 165                 throw new IllegalStateException("Session is closed");\r
 166         }\r
 167     }\r
 168 \r
 169     private class NotificationProviderSessionImpl extends\r
 170             NotificationConsumerSessionImpl implements\r
 171             NotificationProviderService {\r
 172 \r
 173         @Override\r
 174         public void sendNotification(CompositeNode notification) {\r
 175             checkSessionState();\r
 176             if (notification == null)\r
 177                 throw new IllegalArgumentException(\r
 178                         "Notification must not be null.");\r
 179             NotificationModule.this.sendNotification(notification);\r
 180         }\r
 181     }\r
 182 \r
 183     @Override\r
 184     public Set<Class<? extends ProviderFunctionality>> getSupportedProviderFunctionality() {\r
 185         return Collections.emptySet();\r
 186     }\r
 187 }\r