From 6d3cd7698707957a9a240102f174a3be40e8ddf2 2013-02-05 14:57:47 From: Thomas Kluyver Date: 2013-02-05 14:57:47 Subject: [PATCH] Remove unused utils.notification --- diff --git a/IPython/utils/notification.py b/IPython/utils/notification.py deleted file mode 100644 index 506356c..0000000 --- a/IPython/utils/notification.py +++ /dev/null @@ -1,142 +0,0 @@ -# encoding: utf-8 -""" -The IPython Core Notification Center. - -See docs/source/development/notification_blueprint.txt for an overview of the -notification module. - -Authors: - -* Barry Wark -* Brian Granger -""" - -#----------------------------------------------------------------------------- -# Copyright (C) 2008-2011 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. -#----------------------------------------------------------------------------- - -#----------------------------------------------------------------------------- -# Code -#----------------------------------------------------------------------------- - - -class NotificationError(Exception): - pass - - -class NotificationCenter(object): - """Synchronous notification center. - - Examples - -------- - Here is a simple example of how to use this:: - - import IPython.util.notification as notification - def callback(ntype, theSender, args={}): - print ntype,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, ntype, sender, *args, **kwargs): - """Post notification to all registered observers. - - The registered callback will be called as:: - - callback(ntype, sender, *args, **kwargs) - - Parameters - ---------- - ntype : hashable - The notification type. - sender : hashable - The object sending the notification. - *args : tuple - The positional arguments to be passed to the callback. - **kwargs : dict - The keyword argument to be passed to the callback. - - Notes - ----- - * If no registered observers, performance is O(1). - * Notificaiton order is undefined. - * Notifications are posted synchronously. - """ - - if(ntype==None or sender==None): - raise NotificationError( - "Notification type and sender are required.") - - # If there are no registered observers for the type/sender pair - if((ntype 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(ntype, sender): - o(ntype, sender, *args, **kwargs) - - def _observers_for_notification(self, ntype, sender): - """Find all registered observers that should recieve notification""" - - keys = ( - (ntype,sender), - (ntype, 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, ntype, 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 arguments passed - to post_notification. - - Parameters - ---------- - callback : callable - The callable that will be called by :meth:`post_notification` - as ``callback(ntype, sender, *args, **kwargs)`` - ntype : hashable - The notification type. If None, all notifications from sender - will be posted. - sender : hashable - The notification sender. If None, all notifications of ntype - will be posted. - """ - assert(callback != None) - self.registered_types.add(ntype) - self.registered_senders.add(sender) - self.observers.setdefault((ntype,sender), set()).add(callback) - - def remove_all_observers(self): - """Removes all observers from this notification center""" - - self._init_observers() - - - -shared_center = NotificationCenter() diff --git a/IPython/utils/tests/test_notification.py b/IPython/utils/tests/test_notification.py deleted file mode 100644 index 7a58d9e..0000000 --- a/IPython/utils/tests/test_notification.py +++ /dev/null @@ -1,150 +0,0 @@ -# encoding: utf-8 - -"""This file contains unittests for the notification.py module.""" - -#----------------------------------------------------------------------------- -# Copyright (C) 2008-2011 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. -#----------------------------------------------------------------------------- - -#----------------------------------------------------------------------------- -# Imports -#----------------------------------------------------------------------------- - -import unittest - -from IPython.utils.notification import shared_center - -#----------------------------------------------------------------------------- -# Support Classes -#----------------------------------------------------------------------------- - - -class Observer(object): - - def __init__(self, expected_ntype, expected_sender, - center=shared_center, *args, **kwargs): - super(Observer, self).__init__() - self.expected_ntype = expected_ntype - self.expected_sender = expected_sender - self.expected_args = args - self.expected_kwargs = kwargs - self.recieved = False - center.add_observer(self.callback, - self.expected_ntype, - self.expected_sender) - - def callback(self, ntype, sender, *args, **kwargs): - assert(ntype == self.expected_ntype or - self.expected_ntype == None) - assert(sender == self.expected_sender or - self.expected_sender == None) - assert(args == self.expected_args) - assert(kwargs == self.expected_kwargs) - self.recieved = True - - def verify(self): - assert(self.recieved) - - def reset(self): - self.recieved = False - - -class Notifier(object): - - def __init__(self, ntype, **kwargs): - super(Notifier, self).__init__() - self.ntype = ntype - self.kwargs = kwargs - - def post(self, center=shared_center): - - center.post_notification(self.ntype, self, - **self.kwargs) - - -#----------------------------------------------------------------------------- -# Tests -#----------------------------------------------------------------------------- - - -class NotificationTests(unittest.TestCase): - - def tearDown(self): - shared_center.remove_all_observers() - - def test_notification_delivered(self): - """Test that notifications are delivered""" - - expected_ntype = 'EXPECTED_TYPE' - sender = Notifier(expected_ntype) - observer = Observer(expected_ntype, sender) - - sender.post() - observer.verify() - - def test_type_specificity(self): - """Test that observers are registered by type""" - - expected_ntype = 1 - unexpected_ntype = "UNEXPECTED_TYPE" - sender = Notifier(expected_ntype) - unexpected_sender = Notifier(unexpected_ntype) - observer = Observer(expected_ntype, sender) - - sender.post() - unexpected_sender.post() - observer.verify() - - def test_sender_specificity(self): - """Test that observers are registered by sender""" - - expected_ntype = "EXPECTED_TYPE" - sender1 = Notifier(expected_ntype) - sender2 = Notifier(expected_ntype) - observer = Observer(expected_ntype, sender1) - - sender1.post() - sender2.post() - - observer.verify() - - def test_remove_all_observers(self): - """White-box test for remove_all_observers""" - - for i in xrange(10): - Observer('TYPE', None, center=shared_center) - - self.assertTrue(len(shared_center.observers[('TYPE',None)]) >= 10, - "observers registered") - - shared_center.remove_all_observers() - self.assertTrue(len(shared_center.observers) == 0, "observers removed") - - def test_any_sender(self): - expected_ntype = "EXPECTED_TYPE" - sender1 = Notifier(expected_ntype) - sender2 = Notifier(expected_ntype) - observer = Observer(expected_ntype, None) - - sender1.post() - observer.verify() - - observer.reset() - sender2.post() - observer.verify() - - def test_post_performance(self): - """Test that post_notification, even with many registered irrelevant - observers is fast""" - - for i in xrange(10): - Observer("UNRELATED_TYPE", None) - - o = Observer('EXPECTED_TYPE', None) - shared_center.post_notification('EXPECTED_TYPE', self) - o.verify() - -