1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
# 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()
|
