IPython/kernel/core/notification.py

# encoding: utf-8

"""The IPython Core Notification Center.

See docs/source/development/notification_blueprint.txt for an overview of the
notification module.
"""

__docformat__ = "restructuredtext en"
                                                                              
#-----------------------------------------------------------------------------
#  Copyright (C) 2008  The IPython Development Team                           
#                                                                             
#  Distributed under the terms of the BSD License.  The full license is in    
#  the file COPYING, distributed as part of this software.                    
#-----------------------------------------------------------------------------

# Tell nose to skip the testing of this module
__test__ = {}

class NotificationCenter(object):
    """Synchronous notification center
    
    Example
    -------
    >>> import IPython.kernel.core.notification as notification
    >>> def callback(theType, theSender, args={}):
    ...     print theType,theSender,args
    ...     
    >>> notification.sharedCenter.add_observer(callback, 'NOTIFICATION_TYPE', None)
    >>> notification.sharedCenter.post_notification('NOTIFICATION_TYPE', object()) # doctest:+ELLIPSIS
    NOTIFICATION_TYPE ...
        
    """
    def __init__(self):
        super(NotificationCenter, self).__init__()
        self._init_observers()
    
    
    def _init_observers(self):
        """Initialize observer storage"""
        
        self.registered_types = set() #set of types that are observed
        self.registered_senders = set() #set of senders that are observed
        self.observers = {} #map (type,sender) => callback (callable)
    
    
    def post_notification(self, theType, sender, **kwargs):
        """Post notification (type,sender,**kwargs) to all registered
        observers. 
        
        Implementation
        --------------
        * If no registered observers, performance is O(1).
        * Notificaiton order is undefined.
        * Notifications are posted synchronously.
        """
        
        if(theType==None or sender==None):
            raise Exception("NotificationCenter.post_notification requires \
                type and sender.")
        
        # If there are no registered observers for the type/sender pair
        if((theType not in self.registered_types and 
                None not in self.registered_types) or
            (sender not in self.registered_senders and 
                None not in self.registered_senders)):
            return
        
        for o in self._observers_for_notification(theType, sender):
            o(theType, sender, args=kwargs)
    
        
    def _observers_for_notification(self, theType, sender):
        """Find all registered observers that should recieve notification"""
        
        keys = (
                    (theType,sender),
                    (theType, None),
                    (None, sender),  
                    (None,None)
                )
        
        
        obs = set()
        for k in keys:
            obs.update(self.observers.get(k, set()))
        
        return obs
    
    
    def add_observer(self, callback, theType, sender):
        """Add an observer callback to this notification center.
        
        The given callback will be called upon posting of notifications of
        the given type/sender and will receive any additional kwargs passed
        to post_notification.
        
        Parameters
        ----------
        observerCallback : callable
            Callable. Must take at least two arguments::
                observerCallback(type, sender, args={})
        
        theType : hashable
            The notification type. If None, all notifications from sender
            will be posted.
        
        sender : hashable
            The notification sender. If None, all notifications of theType
            will be posted.
        """
        assert(callback != None)
        self.registered_types.add(theType)
        self.registered_senders.add(sender)
        self.observers.setdefault((theType,sender), set()).add(callback)
    
    def remove_all_observers(self):
        """Removes all observers from this notification center"""
        
        self._init_observers()
    


sharedCenter = NotificationCenter()