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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
|
# Copyright (C) 2004 Mike Wray <mike.wray@hp.com>
"""Simple publish/subscribe event server.
"""
import string
from twisted.internet import reactor
# subscribe a.b.c h: map a.b.c -> h
# subscribe a.b.* h: map a.b.* -> h
# subscribe a.b.? h: map a.b.? -> h
#
# for event a.b.c.d:
#
# lookup a.b.c.d, call handlers
#
# lookup a.b.c.?, call handlers
#
# lookup a.b.c.d.*, call handlers
# lookup a.b.c.*, call handlers
# lookup a.b.*, call handlers
# lookup a.*, call handlers
# lookup *, call handlers
# a.b.c.d = (a b c d)
# a.b.c.? = (a b c _)
# a.b.c.* = (a b c . _)
class EventServer:
DOT = '.'
QUERY = '?'
DOT_QUERY = DOT + QUERY
STAR = '*'
DOT_STAR = DOT + STAR
def __init__(self, run=0):
self.handlers = {}
self.run = run
self.queue = []
def start(self):
"""Enable event handling. Sends any queued events.
"""
self.run = 1
for (e,v) in self.queue:
self.inject(e, v)
self.queue = []
def stop(self):
"""Suspend event handling. Events injected while suspended
are queued until we are started again.
"""
self.run = 0
def subscribe(self, event, handler):
"""Subscribe to an event. For example 'a.b.c.d'.
A subcription like 'a.b.c.?' ending in '?' matches any value
for the '?'. A subscription like 'a.b.c.*' ending in '*' matches
any event type with the same prefix, 'a.b.c' in this case.
event event name
handler event handler fn(event, val)
"""
hl = self.handlers.get(event)
if hl is None:
self.handlers[event] = [handler]
else:
hl.append(handler)
def unsubscribe_all(self, event=None):
"""Unsubscribe all handlers for a given event, or all handlers.
event event (optional)
"""
if event == None:
self.handlers.clear()
elif event in self.handlers:
del self.handlers[event]
def unsubscribe(self, event, handler):
"""Unsubscribe a given event and handler.
event event
handler handler
"""
hl = self.handlers.get(event)
if hl is None:
return
if handler in hl:
hl.remove(handler)
def inject(self, event, val, async=1):
"""Inject an event. Handlers for it are called if running, otherwise
it is queued.
event event type
val event value
"""
if self.run:
if async:
reactor.callLater(0, self.call_handlers, event, val)
else:
self.notify_handlers(event, val)
else:
self.queue.append( (event, val) )
def call_handlers(self, event, val):
"""Internal method to call event handlers.
"""
#print ">event", event, val
self.call_event_handlers(event, event, val)
self.call_query_handlers(event, val)
self.call_star_handlers(event, val)
def call_event_handlers(self, key, event, val):
"""Call the handlers for an event.
It is safe for handlers to subscribe or unsubscribe.
key key for handler list
event event type
val event value
"""
hl = self.handlers.get(key)
if hl is None:
return
# Copy the handler list so that handlers can call
# subscribe/unsubscribe safely - python list iteration
# is not safe against list modification.
for h in hl[:]:
try:
h(event, val)
except:
pass
def call_query_handlers(self, event, val):
"""Call regex handlers for events matching 'event' that end in '?'.
event event type
val event value
"""
dot_idx = event.rfind(self.DOT)
if dot_idx == -1:
self.call_event_handlers(self.QUERY, event, val)
else:
event_query = event[0:dot_idx] + self.DOT_QUERY
self.call_event_handlers(event_query, event, val)
def call_star_handlers(self, event, val):
"""Call regex handlers for events matching 'event' that end in '*'.
event event type
val event value
"""
etype = string.split(event, self.DOT)
for i in range(len(etype), 0, -1):
event_star = self.DOT.join(etype[0:i]) + self.DOT_STAR
self.call_event_handlers(event_star, event, val)
self.call_event_handlers(self.STAR, event, val)
def instance():
global inst
try:
inst
except:
inst = EventServer()
inst.start()
return inst
def main():
def sys_star(event, val):
print 'sys_star', event, val
def sys_foo(event, val):
print 'sys_foo', event, val
s.unsubscribe('sys.foo', sys_foo)
def sys_foo2(event, val):
print 'sys_foo2', event, val
def sys_bar(event, val):
print 'sys_bar', event, val
def sys_foo_bar(event, val):
print 'sys_foo_bar', event, val
def foo_bar(event, val):
print 'foo_bar', event, val
s = EventServer()
s.start()
s.subscribe('sys.*', sys_star)
s.subscribe('sys.foo', sys_foo)
s.subscribe('sys.foo', sys_foo2)
s.subscribe('sys.bar', sys_bar)
s.subscribe('sys.foo.bar', sys_foo_bar)
s.subscribe('foo.bar', foo_bar)
s.inject('sys.foo', 'hello')
print
s.inject('sys.bar', 'hello again')
print
s.inject('sys.foo.bar', 'hello again')
print
s.inject('foo.bar', 'hello again')
print
s.inject('foo', 'hello again')
print
s.start()
s.unsubscribe('sys.*', sys_star)
s.unsubscribe_all('sys.*')
s.inject('sys.foo', 'hello')
if __name__ == "__main__":
main()
|
