FilesExpand file tree

 master

/

notification.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

142 lines (108 loc) · 4.47 KB

 master

/

notification.py

Copy path

Top

File metadata and controls

• Code
• Blame

142 lines (108 loc) · 4.47 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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()