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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
# (C) Copyright 2005-2023 Enthought, Inc., Austin, TX
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in LICENSE.txt and may be redistributed only under
# the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
#
# Thanks for using Enthought open source!
def add_or_remove_notifiers(
*, object, graph, handler, target, dispatcher, remove):
""" Add/Remove notifiers on objects following the description on an
ObserverGraph.
All nodes in ``ObserverGraph`` are required to be instances of
``IObserver``. The interface of ``IObserver`` supports this function.
Parameters
----------
object : object
An object to be observed.
graph : ObserverGraph
A graph describing what and how extended traits are being observed.
All nodes must be ``IObserver``.
handler : callable(event)
User-defined callable to handle change events.
``event`` is an object representing the change.
Its type and content depends on the change.
target : Any
An object for defining the context of the user's handler notifier.
This is typically an instance of HasTraits seen by the user as the
"owner" of the observer.
dispatcher : callable(callable, event)
Callable for dispatching the user-defined handler, e.g. dispatching
callback on a different thread.
remove : boolean
If true, notifiers are being removed.
Raises
------
NotiferNotFound
Raised when notifier cannot be found for removal.
"""
callable_ = _AddOrRemoveNotifier(
object=object,
graph=graph,
handler=handler,
target=target,
dispatcher=dispatcher,
remove=remove,
)
callable_()
class _AddOrRemoveNotifier:
""" Callable for adding or removing notifiers.
See ``add_or_remove_notifiers`` for the input parameters.
"""
def __init__(self, *, object, graph, handler, target, dispatcher, remove):
self.object = object
self.graph = graph
self.handler = handler
self.target = target
self.dispatcher = dispatcher
self.remove = remove
# list of (notifier, observable)
self._processed = []
def __call__(self):
""" Main function for adding/removing notifiers.
"""
# The order of events does not matter as they are independent of each
# other.
steps = [
self._add_or_remove_notifiers,
self._add_or_remove_maintainers,
self._add_or_remove_children_notifiers,
self._add_or_remove_extra_graphs,
]
# Not quite the complete reversal, as trees are still walked from
# root to leaves.
if self.remove:
steps = steps[::-1]
try:
for step in steps:
step()
except Exception:
# Undo and then reraise
while self._processed:
notifier, observable = self._processed.pop()
if self.remove:
notifier.add_to(observable)
else:
notifier.remove_from(observable)
raise
else:
self._processed.clear()
def _add_or_remove_extra_graphs(self):
""" Add or remove additional ObserverGraph contributed by the root
observer. e.g. for handing trait_added event.
"""
for extra_graph in self.graph.node.iter_extra_graphs(self.graph):
add_or_remove_notifiers(
object=self.object,
graph=extra_graph,
handler=self.handler,
target=self.target,
dispatcher=self.dispatcher,
remove=self.remove,
)
def _add_or_remove_children_notifiers(self):
""" Recursively add or remove notifiers for the children ObserverGraph.
"""
for child_graph in self.graph.children:
for next_object in self.graph.node.iter_objects(self.object):
add_or_remove_notifiers(
object=next_object,
graph=child_graph,
handler=self.handler,
target=self.target,
dispatcher=self.dispatcher,
remove=self.remove,
)
def _add_or_remove_maintainers(self):
""" Add or remove notifiers for maintaining children notifiers when
the objects being observed by the root observer change.
"""
for observable in self.graph.node.iter_observables(self.object):
for child_graph in self.graph.children:
change_notifier = self.graph.node.get_maintainer(
graph=child_graph,
handler=self.handler,
target=self.target,
dispatcher=self.dispatcher,
)
if self.remove:
change_notifier.remove_from(observable)
else:
change_notifier.add_to(observable)
self._processed.append((change_notifier, observable))
def _add_or_remove_notifiers(self):
""" Add or remove user notifiers for the objects observed by the root
observer.
"""
if not self.graph.node.notify:
return
for observable in self.graph.node.iter_observables(self.object):
notifier = self.graph.node.get_notifier(
handler=self.handler,
target=self.target,
dispatcher=self.dispatcher,
)
if self.remove:
notifier.remove_from(observable)
else:
notifier.add_to(observable)
self._processed.append((notifier, observable))
|
