opendaylight/sal/yang-prototype/sal/sal-broker-impl/src/main/java/org/opendaylight/controller/sal/core/impl/notify/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.impl.BrokerServiceImpl;\r
  26 import org.opendaylight.controller.sal.core.spi.BrokerModule;\r
  27 import org.opendaylight.controller.yang.common.QName;\r
  28 import org.opendaylight.controller.yang.data.api.CompositeNode;\r
  29 import org.slf4j.Logger;\r
  30 import org.slf4j.LoggerFactory;\r
  31 \r
  32 import com.google.common.collect.HashMultimap;\r
  33 import com.google.common.collect.ImmutableSet;\r
  34 import com.google.common.collect.Multimap;\r
  35 \r
  36 public class NotificationModule implements BrokerModule {\r
  37     private static Logger log = LoggerFactory\r
  38             .getLogger(NotificationModule.class);\r
  39 \r
  40     private Multimap<QName, NotificationListener> listeners = HashMultimap\r
  41             .create();\r
  42 \r
  43     private static final Set<Class<? extends BrokerService>> providedServices = ImmutableSet\r
  44             .of((Class<? extends BrokerService>) NotificationService.class,\r
  45                     NotificationProviderService.class);\r
  46 \r
  47     @Override\r
  48     public Set<Class<? extends BrokerService>> getProvidedServices() {\r
  49         return providedServices;\r
  50     }\r
  51 \r
  52     @Override\r
  53     public Set<Class<? extends ConsumerFunctionality>> getSupportedConsumerFunctionality() {\r
  54         // FIXME Refactor\r
  55         Set<Class<? extends ConsumerFunctionality>> ret = new HashSet<Class<? extends ConsumerFunctionality>>();\r
  56         ret.add(NotificationListener.class);\r
  57         return ret;\r
  58     }\r
  59 \r
  60     @Override\r
  61     public <T extends BrokerService> T getServiceForSession(Class<T> service,\r
  62             ConsumerSession session) {\r
  63         if (NotificationProviderService.class.equals(service)\r
  64                 && session instanceof ProviderSession) {\r
  65             @SuppressWarnings("unchecked")\r
  66             T ret = (T) newNotificationProviderService(session);\r
  67             return ret;\r
  68         } else if (NotificationService.class.equals(service)) {\r
  69 \r
  70             @SuppressWarnings("unchecked")\r
  71             T ret = (T) newNotificationConsumerService(session);\r
  72             return ret;\r
  73         }\r
  74 \r
  75         throw new IllegalArgumentException(\r
  76                 "The requested session-specific service is not provided by this module.");\r
  77     }\r
  78 \r
  79     private void sendNotification(CompositeNode notification) {\r
  80         QName type = notification.getNodeType();\r
  81         Collection<NotificationListener> toNotify = listeners.get(type);\r
  82         log.info("Publishing notification " + type);\r
  83 \r
  84         if (toNotify == null) {\r
  85             // No listeners were registered - returns.\r
  86             return;\r
  87         }\r
  88 \r
  89         for (NotificationListener listener : toNotify) {\r
  90             try {\r
  91                 // FIXME: ensure that notification is immutable\r
  92                 listener.onNotification(notification);\r
  93             } catch (Exception e) {\r
  94                 log.error("Uncaught exception in NotificationListener", e);\r
  95             }\r
  96         }\r
  97 \r
  98     }\r
  99 \r
 100     private NotificationService newNotificationConsumerService(\r
 101             ConsumerSession session) {\r
 102         return new NotificationConsumerSessionImpl();\r
 103     }\r
 104 \r
 105     private NotificationProviderService newNotificationProviderService(\r
 106             ConsumerSession session) {\r
 107         return new NotificationProviderSessionImpl();\r
 108     }\r
 109 \r
 110     private class NotificationConsumerSessionImpl extends BrokerServiceImpl\r
 111             implements NotificationService {\r
 112 \r
 113         private Multimap<QName, NotificationListener> consumerListeners = HashMultimap\r
 114                 .create();\r
 115         private boolean closed = false;\r
 116 \r
 117         @Override\r
 118         public void addNotificationListener(QName notification,\r
 119                 NotificationListener listener) {\r
 120             checkSessionState();\r
 121             if (notification == null) {\r
 122                 throw new IllegalArgumentException(\r
 123                         "Notification type must not be null.");\r
 124             }\r
 125             if (listener == null) {\r
 126                 throw new IllegalArgumentException("Listener must not be null.");\r
 127             }\r
 128 \r
 129             consumerListeners.put(notification, listener);\r
 130             listeners.put(notification, listener);\r
 131             log.info("Registered listener for notification: " + notification);\r
 132         }\r
 133 \r
 134         @Override\r
 135         public void removeNotificationListener(QName notification,\r
 136                 NotificationListener listener) {\r
 137             checkSessionState();\r
 138             if (notification == null) {\r
 139                 throw new IllegalArgumentException(\r
 140                         "Notification type must not be null.");\r
 141             }\r
 142             if (listener == null) {\r
 143                 throw new IllegalArgumentException("Listener must not be null.");\r
 144             }\r
 145             consumerListeners.remove(notification, listener);\r
 146             listeners.remove(notification, listener);\r
 147         }\r
 148 \r
 149         @Override\r
 150         public void closeSession() {\r
 151             closed = true;\r
 152             Map<QName, Collection<NotificationListener>> toRemove = consumerListeners\r
 153                     .asMap();\r
 154             for (Entry<QName, Collection<NotificationListener>> entry : toRemove\r
 155                     .entrySet()) {\r
 156                 listeners.remove(entry.getKey(), entry.getValue());\r
 157             }\r
 158         }\r
 159 \r
 160         protected void checkSessionState() {\r
 161             if (closed)\r
 162                 throw new IllegalStateException("Session is closed");\r
 163         }\r
 164     }\r
 165 \r
 166     private class NotificationProviderSessionImpl extends\r
 167             NotificationConsumerSessionImpl implements\r
 168             NotificationProviderService {\r
 169 \r
 170         @Override\r
 171         public void sendNotification(CompositeNode notification) {\r
 172             checkSessionState();\r
 173             if (notification == null)\r
 174                 throw new IllegalArgumentException(\r
 175                         "Notification must not be null.");\r
 176             NotificationModule.this.sendNotification(notification);\r
 177         }\r
 178     }\r
 179 \r
 180     @Override\r
 181     public Set<Class<? extends ProviderFunctionality>> getSupportedProviderFunctionality() {\r
 182         return Collections.emptySet();\r
 183     }\r
 184 }\r