pubsub.py
47.9 KB
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
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
#---------------------------------------------------------------------------
"""
This module provides a publish-subscribe component that allows
listeners to subcribe to messages of a given topic. Contrary to the
original wxPython.lib.pubsub module (which it is based on), it uses
weak referencing to the subscribers so the lifetime of subscribers
is not affected by Publisher. Also, callable objects can be used in
addition to functions and bound methods. See Publisher class docs for
more details.
Thanks to Robb Shecter and Robin Dunn for having provided
the basis for this module (which now shares most of the concepts but
very little design or implementation with the original
wxPython.lib.pubsub).
The publisher is a singleton instance of the PublisherClass class. You
access the instance via the Publisher object available from the module::
from wx.lib.pubsub import Publisher
Publisher().subscribe(...)
Publisher().sendMessage(...)
...
:Author: Oliver Schoenborn
:Since: Apr 2004
:Version: $Id: pubsub.py 69063 2011-09-11 18:19:32Z CJP $
:Copyright: \(c) 2004 Oliver Schoenborn
:License: wxWidgets
"""
_implNotes = """
Implementation notes
--------------------
In class Publisher, I represent the topics-listener set as a tree
where each node is a topic, and contains a list of listeners of that
topic, and a dictionary of subtopics of that topic. When the Publisher
is told to send a message for a given topic, it traverses the tree
down to the topic for which a message is being generated, all
listeners on the way get sent the message.
Publisher currently uses a weak listener topic tree to store the
topics for each listener, and if a listener dies before being
unsubscribed, the tree is notified, and the tree eliminates the
listener from itself.
Ideally, _TopicTreeNode would be a generic _TreeNode with named
subnodes, and _TopicTreeRoot would be a generic _Tree with named
nodes, and Publisher would store listeners in each node and a topic
tuple would be converted to a path in the tree. This would lead to a
much cleaner separation of concerns. But time is over, time to move on.
"""
#---------------------------------------------------------------------------
# for function and method parameter counting:
from types import InstanceType
from inspect import getargspec, ismethod, isfunction
# for weakly bound methods:
from new import instancemethod as InstanceMethod
from weakref import ref as WeakRef
# -----------------------------------------------------------------------------
def _isbound(method):
"""Return true if method is a bound method, false otherwise"""
assert ismethod(method)
return method.im_self is not None
def _paramMinCountFunc(function):
"""Given a function, return pair (min,d) where min is minimum # of
args required, and d is number of default arguments."""
assert isfunction(function)
(args, va, kwa, dflt) = getargspec(function)
lenDef = len(dflt or ())
lenArgs = len(args or ())
lenVA = int(va is not None)
return (lenArgs - lenDef + lenVA, lenDef)
def _paramMinCount(callableObject):
"""
Given a callable object (function, method or callable instance),
return pair (min,d) where min is minimum # of args required, and d
is number of default arguments. The 'self' parameter, in the case
of methods, is not counted.
"""
if type(callableObject) is InstanceType:
min, d = _paramMinCountFunc(callableObject.__call__.im_func)
return min-1, d
elif ismethod(callableObject):
min, d = _paramMinCountFunc(callableObject.im_func)
return min-1, d
elif isfunction(callableObject):
return _paramMinCountFunc(callableObject)
else:
raise 'Cannot determine type of callable: '+repr(callableObject)
def _tupleize(items):
"""Convert items to tuple if not already one,
so items must be a list, tuple or non-sequence"""
if isinstance(items, list):
raise TypeError, 'Not allowed to tuple-ize a list'
elif isinstance(items, (str, unicode)) and items.find('.') != -1:
items = tuple(items.split('.'))
elif not isinstance(items, tuple):
items = (items,)
return items
def _getCallableName(callable):
"""Get name for a callable, ie function, bound
method or callable instance"""
if ismethod(callable):
return '%s.%s ' % (callable.im_self, callable.im_func.func_name)
elif isfunction(callable):
return '%s ' % callable.__name__
else:
return '%s ' % callable
def _removeItem(item, fromList):
"""Attempt to remove item from fromList, return true
if successful, false otherwise."""
try:
fromList.remove(item)
return True
except ValueError:
return False
# -----------------------------------------------------------------------------
class _WeakMethod:
"""Represent a weak bound method, i.e. a method doesn't keep alive the
object that it is bound to. It uses WeakRef which, used on its own,
produces weak methods that are dead on creation, not very useful.
Typically, you will use the getRef() function instead of using
this class directly. """
def __init__(self, method, notifyDead = None):
"""The method must be bound. notifyDead will be called when
object that method is bound to dies. """
assert ismethod(method)
if method.im_self is None:
raise ValueError, "We need a bound method!"
if notifyDead is None:
self.objRef = WeakRef(method.im_self)
else:
self.objRef = WeakRef(method.im_self, notifyDead)
self.fun = method.im_func
self.cls = method.im_class
def __call__(self):
"""Returns a new.instancemethod if object for method still alive.
Otherwise return None. Note that instancemethod causes a
strong reference to object to be created, so shouldn't save
the return value of this call. Note also that this __call__
is required only for compatibility with WeakRef.ref(), otherwise
there would be more efficient ways of providing this functionality."""
if self.objRef() is None:
return None
else:
return InstanceMethod(self.fun, self.objRef(), self.cls)
def __eq__(self, method2):
"""Two WeakMethod objects compare equal if they refer to the same method
of the same instance. Thanks to Josiah Carlson for patch and clarifications
on how dict uses eq/cmp and hashing. """
if not isinstance(method2, _WeakMethod):
return False
return self.fun is method2.fun \
and self.objRef() is method2.objRef() \
and self.objRef() is not None
def __hash__(self):
"""Hash is an optimization for dict searches, it need not
return different numbers for every different object. Some objects
are not hashable (eg objects of classes derived from dict) so no
hash(objRef()) in there, and hash(self.cls) would only be useful
in the rare case where instance method was rebound. """
return hash(self.fun)
def __repr__(self):
dead = ''
if self.objRef() is None:
dead = '; DEAD'
obj = '<%s at %s%s>' % (self.__class__, id(self), dead)
return obj
def refs(self, weakRef):
"""Return true if we are storing same object referred to by weakRef."""
return self.objRef == weakRef
def _getWeakRef(obj, notifyDead=None):
"""Get a weak reference to obj. If obj is a bound method, a _WeakMethod
object, that behaves like a WeakRef, is returned, if it is
anything else a WeakRef is returned. If obj is an unbound method,
a ValueError will be raised."""
if ismethod(obj):
createRef = _WeakMethod
else:
createRef = WeakRef
if notifyDead is None:
return createRef(obj)
else:
return createRef(obj, notifyDead)
# -----------------------------------------------------------------------------
def getStrAllTopics():
"""Function to call if, for whatever reason, you need to know
explicitely what is the string to use to indicate 'all topics'."""
return ''
# alias, easier to see where used
ALL_TOPICS = getStrAllTopics()
# -----------------------------------------------------------------------------
class _NodeCallback:
"""Encapsulate a weak reference to a method of a TopicTreeNode
in such a way that the method can be called, if the node is
still alive, but the callback does not *keep* the node alive.
Also, define two methods, preNotify() and noNotify(), which can
be redefined to something else, very useful for testing.
"""
def __init__(self, obj):
self.objRef = _getWeakRef(obj)
def __call__(self, weakCB):
notify = self.objRef()
if notify is not None:
self.preNotify(weakCB)
notify(weakCB)
else:
self.noNotify()
def preNotify(self, dead):
"""'Gets called just before our callback (self.objRef) is called"""
pass
def noNotify(self):
"""Gets called if the TopicTreeNode for this callback is dead"""
pass
class _TopicTreeNode:
"""A node in the topic tree. This contains a list of callables
that are interested in the topic that this node is associated
with, and contains a dictionary of subtopics, whose associated
values are other _TopicTreeNodes. The topic of a node is not stored
in the node, so that the tree can be implemented as a dictionary
rather than a list, for ease of use (and, likely, performance).
Note that it uses _NodeCallback to encapsulate a callback for
when a registered listener dies, possible thanks to WeakRef.
Whenever this callback is called, the onDeadListener() function,
passed in at construction time, is called (unless it is None).
"""
def __init__(self, topicPath, onDeadListenerWeakCB):
self.__subtopics = {}
self.__callables = []
self.__topicPath = topicPath
self.__onDeadListenerWeakCB = onDeadListenerWeakCB
def getPathname(self):
"""The complete node path to us, ie., the topic tuple that would lead to us"""
return self.__topicPath
def createSubtopic(self, subtopic, topicPath):
"""Create a child node for subtopic"""
return self.__subtopics.setdefault(subtopic,
_TopicTreeNode(topicPath, self.__onDeadListenerWeakCB))
def hasSubtopic(self, subtopic):
"""Return true only if topic string is one of subtopics of this node"""
return self.__subtopics.has_key(subtopic)
def getNode(self, subtopic):
"""Return ref to node associated with subtopic"""
return self.__subtopics[subtopic]
def addCallable(self, callable):
"""Add a callable to list of callables for this topic node"""
try:
id = self.__callables.index(_getWeakRef(callable))
return self.__callables[id]
except ValueError:
wrCall = _getWeakRef(callable, _NodeCallback(self.__notifyDead))
self.__callables.append(wrCall)
return wrCall
def getCallables(self):
"""Get callables associated with this topic node"""
return [cb() for cb in self.__callables if cb() is not None]
def hasCallable(self, callable):
"""Return true if callable in this node"""
try:
self.__callables.index(_getWeakRef(callable))
return True
except ValueError:
return False
def sendMessage(self, message):
"""Send a message to our callables"""
deliveryCount = 0
for cb in self.__callables:
listener = cb()
if listener is not None:
listener(message)
deliveryCount += 1
return deliveryCount
def removeCallable(self, callable):
"""Remove weak callable from our node (and return True).
Does nothing if not here (and returns False)."""
try:
self.__callables.remove(_getWeakRef(callable))
return True
except ValueError:
return False
def clearCallables(self):
"""Abandon list of callables to caller. We no longer have
any callables after this method is called."""
tmpList = [cb for cb in self.__callables if cb() is not None]
self.__callables = []
return tmpList
def __notifyDead(self, dead):
"""Gets called when a listener dies, thanks to WeakRef"""
#print 'TreeNODE', `self`, 'received death certificate for ', dead
self.__cleanupDead()
if self.__onDeadListenerWeakCB is not None:
cb = self.__onDeadListenerWeakCB()
if cb is not None:
cb(dead)
def __cleanupDead(self):
"""Remove all dead objects from list of callables"""
self.__callables = [cb for cb in self.__callables if cb() is not None]
def __str__(self):
"""Print us in a not-so-friendly, but readable way, good for debugging."""
strVal = []
for callable in self.getCallables():
strVal.append(_getCallableName(callable))
for topic, node in self.__subtopics.iteritems():
strVal.append(' (%s: %s)' %(topic, node))
return ''.join(strVal)
class _TopicTreeRoot(_TopicTreeNode):
"""
The root of the tree knows how to access other node of the
tree and is the gateway of the tree user to the tree nodes.
It can create topics, and and remove callbacks, etc.
For efficiency, it stores a dictionary of listener-topics,
so that unsubscribing a listener just requires finding the
topics associated to a listener, and finding the corresponding
nodes of the tree. Without it, unsubscribing would require
that we search the whole tree for all nodes that contain
given listener. Since Publisher is a singleton, it will
contain all topics in the system so it is likely to be a large
tree. However, it is possible that in some runs, unsubscribe()
is called very little by the user, in which case most unsubscriptions
are automatic, ie caused by the listeners dying. In this case,
a flag is set to indicate that the dictionary should be cleaned up
at the next opportunity. This is not necessary, it is just an
optimization.
"""
def __init__(self):
self.__callbackDict = {}
self.__callbackDictCleanup = 0
# all child nodes will call our __rootNotifyDead method
# when one of their registered listeners dies
_TopicTreeNode.__init__(self, (ALL_TOPICS,),
_getWeakRef(self.__rootNotifyDead))
def addTopic(self, topic, listener):
"""Add topic to tree if doesnt exist, and add listener to topic node"""
assert isinstance(topic, tuple)
topicNode = self.__getTreeNode(topic, make=True)
weakCB = topicNode.addCallable(listener)
assert topicNode.hasCallable(listener)
theList = self.__callbackDict.setdefault(weakCB, [])
assert self.__callbackDict.has_key(weakCB)
# add it only if we don't already have it
try:
weakTopicNode = WeakRef(topicNode)
theList.index(weakTopicNode)
except ValueError:
theList.append(weakTopicNode)
assert self.__callbackDict[weakCB].index(weakTopicNode) >= 0
def getTopics(self, listener):
"""Return the list of topics for given listener"""
weakNodes = self.__callbackDict.get(_getWeakRef(listener), [])
return [weakNode().getPathname() for weakNode in weakNodes
if weakNode() is not None]
def isSubscribed(self, listener, topic=None):
"""Return true if listener is registered for topic specified.
If no topic specified, return true if subscribed to something.
Use topic=getStrAllTopics() to determine if a listener will receive
messages for all topics."""
weakCB = _getWeakRef(listener)
if topic is None:
return self.__callbackDict.has_key(weakCB)
else:
topicPath = _tupleize(topic)
for weakNode in self.__callbackDict[weakCB]:
if topicPath == weakNode().getPathname():
return True
return False
def unsubscribe(self, listener, topicList):
"""Remove listener from given list of topics. If topicList
doesn't have any topics for which listener has subscribed,
nothing happens."""
weakCB = _getWeakRef(listener)
if not self.__callbackDict.has_key(weakCB):
return
cbNodes = self.__callbackDict[weakCB]
if topicList is None:
for weakNode in cbNodes:
weakNode().removeCallable(listener)
del self.__callbackDict[weakCB]
return
for weakNode in cbNodes:
node = weakNode()
if node is not None and node.getPathname() in topicList:
success = node.removeCallable(listener)
assert success == True
cbNodes.remove(weakNode)
assert not self.isSubscribed(listener, node.getPathname())
def unsubAll(self, topicList, onNoSuchTopic):
"""Unsubscribe all listeners registered for any topic in
topicList. If a topic in the list does not exist, and
onNoSuchTopic is not None, a call
to onNoSuchTopic(topic) is done for that topic."""
for topic in topicList:
node = self.__getTreeNode(topic)
if node is not None:
weakCallables = node.clearCallables()
for callable in weakCallables:
weakNodes = self.__callbackDict[callable]
success = _removeItem(WeakRef(node), weakNodes)
assert success == True
if weakNodes == []:
del self.__callbackDict[callable]
elif onNoSuchTopic is not None:
onNoSuchTopic(topic)
def sendMessage(self, topic, message, onTopicNeverCreated):
"""Send a message for given topic to all registered listeners. If
topic doesn't exist, call onTopicNeverCreated(topic)."""
# send to the all-toipcs listeners
deliveryCount = _TopicTreeNode.sendMessage(self, message)
# send to those who listen to given topic or any of its supertopics
node = self
for topicItem in topic:
assert topicItem != ''
if node.hasSubtopic(topicItem):
node = node.getNode(topicItem)
deliveryCount += node.sendMessage(message)
else: # topic never created, don't bother continuing
if onTopicNeverCreated is not None:
onTopicNeverCreated(topic)
break
return deliveryCount
def numListeners(self):
"""Return a pair (live, dead) with count of live and dead listeners in tree"""
dead, live = 0, 0
for cb in self.__callbackDict:
if cb() is None:
dead += 1
else:
live += 1
return live, dead
# clean up the callback dictionary after how many dead listeners
callbackDeadLimit = 10
def __rootNotifyDead(self, dead):
#print 'TreeROOT received death certificate for ', dead
self.__callbackDictCleanup += 1
if self.__callbackDictCleanup > _TopicTreeRoot.callbackDeadLimit:
self.__callbackDictCleanup = 0
oldDict = self.__callbackDict
self.__callbackDict = {}
for weakCB, weakNodes in oldDict.iteritems():
if weakCB() is not None:
self.__callbackDict[weakCB] = weakNodes
def __getTreeNode(self, topic, make=False):
"""Return the tree node for 'topic' from the topic tree. If it
doesnt exist and make=True, create it first."""
# if the all-topics, give root;
if topic == (ALL_TOPICS,):
return self
# not root, so traverse tree
node = self
path = ()
for topicItem in topic:
path += (topicItem,)
if topicItem == ALL_TOPICS:
raise ValueError, 'Topic tuple must not contain ""'
if make:
node = node.createSubtopic(topicItem, path)
elif node.hasSubtopic(topicItem):
node = node.getNode(topicItem)
else:
return None
# done
return node
def printCallbacks(self):
strVal = ['Callbacks:\n']
for listener, weakTopicNodes in self.__callbackDict.iteritems():
topics = [topic() for topic in weakTopicNodes if topic() is not None]
strVal.append(' %s: %s\n' % (_getCallableName(listener()), topics))
return ''.join(strVal)
def __str__(self):
return 'all: %s' % _TopicTreeNode.__str__(self)
# -----------------------------------------------------------------------------
class _SingletonKey: pass
class PublisherClass:
"""
The publish/subscribe manager. It keeps track of which listeners
are interested in which topics (see subscribe()), and sends a
Message for a given topic to listeners that have subscribed to
that topic, with optional user data (see sendMessage()).
The three important concepts for Publisher are:
- listener: a function, bound method or
callable object that can be called with one parameter
(not counting 'self' in the case of methods). The parameter
will be a reference to a Message object. E.g., these listeners
are ok::
class Foo:
def __call__(self, a, b=1): pass # can be called with only one arg
def meth(self, a): pass # takes only one arg
def meth2(self, a=2, b=''): pass # can be called with one arg
def func(a, b=''): pass
Foo foo
Publisher().subscribe(foo) # functor
Publisher().subscribe(foo.meth) # bound method
Publisher().subscribe(foo.meth2) # bound method
Publisher().subscribe(func) # function
The three types of callables all have arguments that allow a call
with only one argument. In every case, the parameter 'a' will contain
the message.
- topic: a single word, a tuple of words, or a string containing a
set of words separated by dots, for example: 'sports.baseball'.
A tuple or a dotted notation string denotes a hierarchy of
topics from most general to least. For example, a listener of
this topic::
('sports','baseball')
would receive messages for these topics::
('sports', 'baseball') # because same
('sports', 'baseball', 'highscores') # because more specific
but not these::
'sports' # because more general
('sports',) # because more general
() or ('') # because only for those listening to 'all' topics
('news') # because different topic
- message: this is an instance of Message, containing the topic for
which the message was sent, and any data the sender specified.
:note: This class is visible to importers of pubsub only as a
Singleton. I.e., every time you execute 'Publisher()', it's
actually the same instance of PublisherClass that is
returned. So to use, just do'Publisher().method()'.
"""
__ALL_TOPICS_TPL = (ALL_TOPICS, )
def __init__(self, singletonKey):
"""Construct a Publisher. This can only be done by the pubsub
module. You just use pubsub.Publisher()."""
if not isinstance(singletonKey, _SingletonKey):
raise invalid_argument("Use Publisher() to get access to singleton")
self.__messageCount = 0
self.__deliveryCount = 0
self.__topicTree = _TopicTreeRoot()
#
# Public API
#
def getDeliveryCount(self):
"""How many listeners have received a message since beginning of run"""
return self.__deliveryCount
def getMessageCount(self):
"""How many times sendMessage() was called since beginning of run"""
return self.__messageCount
def subscribe(self, listener, topic = ALL_TOPICS):
"""
Subscribe listener for given topic. If topic is not specified,
listener will be subscribed for all topics (that listener will
receive a Message for any topic for which a message is generated).
This method may be called multiple times for one listener,
registering it with many topics. It can also be invoked many
times for a particular topic, each time with a different
listener. See the class doc for requirements on listener and
topic.
:note: The listener is held by Publisher() only by *weak*
reference. This means you must ensure you have at
least one strong reference to listener, otherwise it
will be DOA ("dead on arrival"). This is particularly
easy to forget when wrapping a listener method in a
proxy object (e.g. to bind some of its parameters),
e.g.::
class Foo:
def listener(self, event): pass
class Wrapper:
def __init__(self, fun): self.fun = fun
def __call__(self, *args): self.fun(*args)
foo = Foo()
Publisher().subscribe( Wrapper(foo.listener) ) # whoops: DOA!
wrapper = Wrapper(foo.listener)
Publisher().subscribe(wrapper) # good!
:note: Calling this method for the same listener, with two
topics in the same branch of the topic hierarchy, will
cause the listener to be notified twice when a message
for the deepest topic is sent. E.g.
subscribe(listener, 't1') and then subscribe(listener,
('t1','t2')) means that when calling sendMessage('t1'),
listener gets one message, but when calling
sendMessage(('t1','t2')), listener gets message twice.
"""
self.validate(listener)
if topic is None:
raise TypeError, 'Topic must be either a word, tuple of '\
'words, or getStrAllTopics()'
self.__topicTree.addTopic(_tupleize(topic), listener)
def isSubscribed(self, listener, topic=None):
"""Return true if listener has subscribed to topic specified.
If no topic specified, return true if subscribed to something.
Use topic=getStrAllTopics() to determine if a listener will receive
messages for all topics."""
return self.__topicTree.isSubscribed(listener, topic)
def validate(self, listener):
"""Similar to isValid(), but raises a TypeError exception if not valid"""
# check callable
if not callable(listener):
raise TypeError, 'Listener '+`listener`+' must be a '\
'function, bound method or instance.'
# ok, callable, but if method, is it bound:
elif ismethod(listener) and not _isbound(listener):
raise TypeError, 'Listener '+`listener`+\
' is a method but it is unbound!'
# check that it takes the right number of parameters
min, d = _paramMinCount(listener)
if min > 1:
raise TypeError, 'Listener '+`listener`+" can't"\
' require more than one parameter!'
if min <= 0 and d == 0:
raise TypeError, 'Listener '+`listener`+' lacking arguments!'
assert (min == 0 and d>0) or (min == 1)
def isValid(self, listener):
"""Return true only if listener will be able to subscribe to
Publisher."""
try:
self.validate(listener)
return True
except TypeError:
return False
def unsubAll(self, topics=None, onNoSuchTopic=None):
"""Unsubscribe all listeners subscribed for topics. Topics can
be a single topic (string or tuple) or a list of topics (ie
list containing strings and/or tuples). If topics is not
specified, all listeners for all topics will be unsubscribed,
ie. the Publisher singleton will have no topics and no listeners
left. If onNoSuchTopic is given, it will be called as
onNoSuchTopic(topic) for each topic that is unknown.
"""
if topics is None:
del self.__topicTree
self.__topicTree = _TopicTreeRoot()
return
# make sure every topics are in tuple form
if isinstance(topics, list):
topicList = [_tupleize(x) for x in topics]
else:
topicList = [_tupleize(topics)]
# unsub every listener of topics
self.__topicTree.unsubAll(topicList, onNoSuchTopic)
def unsubscribe(self, listener, topics=None):
"""Unsubscribe listener. If topics not specified, listener is
completely unsubscribed. Otherwise, it is unsubscribed only
for the topic (the usual tuple) or list of topics (ie a list
of tuples) specified. Nothing happens if listener is not actually
subscribed to any of the topics.
Note that if listener subscribed for two topics (a,b) and (a,c),
then unsubscribing for topic (a) will do nothing. You must
use getAssociatedTopics(listener) and give unsubscribe() the returned
list (or a subset thereof).
"""
self.validate(listener)
topicList = None
if topics is not None:
if isinstance(topics, list):
topicList = [_tupleize(x) for x in topics]
else:
topicList = [_tupleize(topics)]
self.__topicTree.unsubscribe(listener, topicList)
def getAssociatedTopics(self, listener):
"""Return a list of topics the given listener is registered with.
Returns [] if listener never subscribed.
:attention: when using the return of this method to compare to
expected list of topics, remember that topics that are
not in the form of a tuple appear as a one-tuple in
the return. E.g. if you have subscribed a listener to
'topic1' and ('topic2','subtopic2'), this method
returns::
associatedTopics = [('topic1',), ('topic2','subtopic2')]
"""
return self.__topicTree.getTopics(listener)
def sendMessage(self, topic=ALL_TOPICS,
data=None, onTopicNeverCreated=None,
context=None):
"""Send a message for given topic, with optional data, to
subscribed listeners. If topic is not specified, only the
listeners that are interested in all topics will receive message.
The onTopicNeverCreated is an optional callback of your choice that
will be called if the topic given was never created (i.e. it, or
one of its subtopics, was never subscribed to by any listener).
It will be called as onTopicNeverCreated(topic)."""
aTopic = _tupleize(topic)
message = Message(aTopic, data, context=context)
self.__messageCount += 1
# send to those who listen to all topics
self.__deliveryCount += \
self.__topicTree.sendMessage(aTopic, message, onTopicNeverCreated)
#
# Private methods
#
def __call__(self):
"""Allows for singleton"""
return self
def __str__(self):
return str(self.__topicTree)
# Create the Publisher singleton. We prevent users from (inadvertently)
# instantiating more than one object, by requiring a key that is
# accessible only to module. From
# this point forward any calls to Publisher() will invoke the __call__
# of this instance which just returns itself.
#
# The only flaw with this approach is that you can't derive a new
# class from Publisher without jumping through hoops. If this ever
# becomes an issue then a new Singleton implementaion will need to be
# employed.
_key = _SingletonKey()
Publisher = PublisherClass(_key)
#---------------------------------------------------------------------------
class Message:
"""
A simple container object for the two components of a message: the
topic and the user data. An instance of Message is given to your
listener when called by Publisher().sendMessage(topic) (if your
listener callback was registered for that topic).
"""
def __init__(self, topic, data, context=None):
self.topic = topic
self.data = data
self.context = context
Data = property(lambda self: self.data,
lambda self, data: setattr(self, 'data', data))
Context = property(lambda self: self.context,
lambda self, ctx: setattr(self, 'context', value))
Type = property(lambda self: self.topic)
def __str__(self):
return '[Topic: '+`self.topic`+', Data: '+`self.data`+']'
def GetContext(self):
"""Get the context that this message was sent in"""
return self.context
def GetData(self):
"""Return the messages data/value"""
return self.data
def GetType(self):
"""Return the type of message"""
return self.topic
#---------------------------------------------------------------------------
#
# Code for a simple command-line test
#
def test():
def done(funcName):
print '----------- Done %s -----------' % funcName
def testParam():
def testFunc00(): pass
def testFunc21(a,b,c=1): pass
def testFuncA(*args): pass
def testFuncAK(*args,**kwds): pass
def testFuncK(**kwds): pass
class Foo:
def testMeth(self,a,b): pass
def __call__(self, a): pass
class Foo2:
def __call__(self, *args): pass
assert _paramMinCount(testFunc00)==(0,0)
assert _paramMinCount(testFunc21)==(2,1)
assert _paramMinCount(testFuncA) ==(1,0)
assert _paramMinCount(testFuncAK)==(1,0)
assert _paramMinCount(testFuncK) ==(0,0)
foo = Foo()
assert _paramMinCount(Foo.testMeth)==(2,0)
assert _paramMinCount(foo.testMeth)==(2,0)
assert _paramMinCount(foo)==(1,0)
assert _paramMinCount(Foo2())==(1,0)
done('testParam')
testParam()
#------------------------
_NodeCallback.notified = 0
def testPreNotifyNode(self, dead):
_NodeCallback.notified += 1
print 'testPreNotifyNODE heard notification of', `dead`
_NodeCallback.preNotify = testPreNotifyNode
def testTreeNode():
class WS:
def __init__(self, s):
self.s = s
def __call__(self, msg):
print 'WS#', self.s, ' received msg ', msg
def __str__(self):
return self.s
def testPreNotifyRoot(dead):
print 'testPreNotifyROOT heard notification of', `dead`
node = _TopicTreeNode((ALL_TOPICS,), WeakRef(testPreNotifyRoot))
boo, baz, bid = WS('boo'), WS('baz'), WS('bid')
node.addCallable(boo)
node.addCallable(baz)
node.addCallable(boo)
assert node.getCallables() == [boo,baz]
assert node.hasCallable(boo)
node.removeCallable(bid) # no-op
assert node.hasCallable(baz)
assert node.getCallables() == [boo,baz]
node.removeCallable(boo)
assert node.getCallables() == [baz]
assert node.hasCallable(baz)
assert not node.hasCallable(boo)
node.removeCallable(baz)
assert node.getCallables() == []
assert not node.hasCallable(baz)
node2 = node.createSubtopic('st1', ('st1',))
node3 = node.createSubtopic('st2', ('st2',))
cb1, cb2, cb = WS('st1_cb1'), WS('st1_cb2'), WS('st2_cb')
node2.addCallable(cb1)
node2.addCallable(cb2)
node3.addCallable(cb)
node2.createSubtopic('st3', ('st1','st3'))
node2.createSubtopic('st4', ('st1','st4'))
print str(node)
assert str(node) == ' (st1: st1_cb1 st1_cb2 (st4: ) (st3: )) (st2: st2_cb )'
# verify send message, and that a dead listener does not get sent one
delivered = node2.sendMessage('hello')
assert delivered == 2
del cb1
delivered = node2.sendMessage('hello')
assert delivered == 1
assert _NodeCallback.notified == 1
done('testTreeNode')
testTreeNode()
#------------------------
def testValidate():
class Foo:
def __call__(self, a): pass
def fun(self, b): pass
def fun2(self, b=1): pass
def fun3(self, a, b=2): pass
def badFun(self): pass
def badFun2(): pass
def badFun3(self, a, b): pass
server = Publisher()
foo = Foo()
server.validate(foo)
server.validate(foo.fun)
server.validate(foo.fun2)
server.validate(foo.fun3)
assert not server.isValid(foo.badFun)
assert not server.isValid(foo.badFun2)
assert not server.isValid(foo.badFun3)
done('testValidate')
testValidate()
#------------------------
class SimpleListener:
def __init__(self, number):
self.number = number
def __call__(self, message = ''):
print 'Callable #%s got the message "%s"' %(self.number, message)
def notify(self, message):
print '%s.notify() got the message "%s"' %(self.number, message)
def __str__(self):
return "SimpleListener_%s" % self.number
def testSubscribe():
publisher = Publisher()
topic1 = 'politics'
topic2 = ('history','middle age')
topic3 = ('politics','UN')
topic4 = ('politics','NATO')
topic5 = ('politics','NATO','US')
lisnr1 = SimpleListener(1)
lisnr2 = SimpleListener(2)
def func(message, a=1):
print 'Func received message "%s"' % message
lisnr3 = func
lisnr4 = lambda x: 'Lambda received message "%s"' % x
assert not publisher.isSubscribed(lisnr1)
assert not publisher.isSubscribed(lisnr2)
assert not publisher.isSubscribed(lisnr3)
assert not publisher.isSubscribed(lisnr4)
publisher.subscribe(lisnr1, topic1)
assert publisher.getAssociatedTopics(lisnr1) == [(topic1,)]
publisher.subscribe(lisnr1, topic2)
publisher.subscribe(lisnr1, topic1) # do it again, should be no-op
assert publisher.getAssociatedTopics(lisnr1) == [(topic1,),topic2]
publisher.subscribe(lisnr2.notify, topic3)
assert publisher.getAssociatedTopics(lisnr2.notify) == [topic3]
assert publisher.getAssociatedTopics(lisnr1) == [(topic1,),topic2]
publisher.subscribe(lisnr3, topic5)
assert publisher.getAssociatedTopics(lisnr3) == [topic5]
assert publisher.getAssociatedTopics(lisnr2.notify) == [topic3]
assert publisher.getAssociatedTopics(lisnr1) == [(topic1,),topic2]
publisher.subscribe(lisnr4)
print "Publisher tree: ", publisher
assert publisher.isSubscribed(lisnr1)
assert publisher.isSubscribed(lisnr1, topic1)
assert publisher.isSubscribed(lisnr1, topic2)
assert publisher.isSubscribed(lisnr2.notify)
assert publisher.isSubscribed(lisnr3, topic5)
assert publisher.isSubscribed(lisnr4, ALL_TOPICS)
expectTopicTree = 'all: <lambda> (politics: SimpleListener_1 (UN: SimpleListener_2.notify ) (NATO: (US: func ))) (history: (middle age: SimpleListener_1 ))'
print "Publisher tree: ", publisher
assert str(publisher) == expectTopicTree
publisher.unsubscribe(lisnr1, 'booboo') # should do nothing
assert publisher.getAssociatedTopics(lisnr1) == [(topic1,),topic2]
assert publisher.getAssociatedTopics(lisnr2.notify) == [topic3]
assert publisher.getAssociatedTopics(lisnr3) == [topic5]
publisher.unsubscribe(lisnr1, topic1)
assert publisher.getAssociatedTopics(lisnr1) == [topic2]
assert publisher.getAssociatedTopics(lisnr2.notify) == [topic3]
assert publisher.getAssociatedTopics(lisnr3) == [topic5]
publisher.unsubscribe(lisnr1, topic2)
publisher.unsubscribe(lisnr1, topic2)
publisher.unsubscribe(lisnr2.notify, topic3)
publisher.unsubscribe(lisnr3, topic5)
assert publisher.getAssociatedTopics(lisnr1) == []
assert publisher.getAssociatedTopics(lisnr2.notify) == []
assert publisher.getAssociatedTopics(lisnr3) == []
publisher.unsubscribe(lisnr4)
expectTopicTree = 'all: (politics: (UN: ) (NATO: (US: ))) (history: (middle age: ))'
print "Publisher tree: ", publisher
assert str(publisher) == expectTopicTree
assert publisher.getDeliveryCount() == 0
assert publisher.getMessageCount() == 0
publisher.unsubAll()
assert str(publisher) == 'all: '
done('testSubscribe')
testSubscribe()
#------------------------
def testUnsubAll():
publisher = Publisher()
topic1 = 'politics'
topic2 = ('history','middle age')
topic3 = ('politics','UN')
topic4 = ('politics','NATO')
topic5 = ('politics','NATO','US')
lisnr1 = SimpleListener(1)
lisnr2 = SimpleListener(2)
def func(message, a=1):
print 'Func received message "%s"' % message
lisnr3 = func
lisnr4 = lambda x: 'Lambda received message "%s"' % x
publisher.subscribe(lisnr1, topic1)
publisher.subscribe(lisnr1, topic2)
publisher.subscribe(lisnr2.notify, topic3)
publisher.subscribe(lisnr3, topic2)
publisher.subscribe(lisnr3, topic5)
publisher.subscribe(lisnr4)
expectTopicTree = 'all: <lambda> (politics: SimpleListener_1 (UN: SimpleListener_2.notify ) (NATO: (US: func ))) (history: (middle age: SimpleListener_1 func ))'
print "Publisher tree: ", publisher
assert str(publisher) == expectTopicTree
publisher.unsubAll(topic1)
assert publisher.getAssociatedTopics(lisnr1) == [topic2]
assert not publisher.isSubscribed(lisnr1, topic1)
publisher.unsubAll(topic2)
print publisher
assert publisher.getAssociatedTopics(lisnr1) == []
assert publisher.getAssociatedTopics(lisnr3) == [topic5]
assert not publisher.isSubscribed(lisnr1)
assert publisher.isSubscribed(lisnr3, topic5)
#print "Publisher tree: ", publisher
expectTopicTree = 'all: <lambda> (politics: (UN: SimpleListener_2.notify ) (NATO: (US: func ))) (history: (middle age: ))'
assert str(publisher) == expectTopicTree
publisher.unsubAll(ALL_TOPICS)
#print "Publisher tree: ", publisher
expectTopicTree = 'all: (politics: (UN: SimpleListener_2.notify ) (NATO: (US: func ))) (history: (middle age: ))'
assert str(publisher) == expectTopicTree
publisher.unsubAll()
done('testUnsubAll')
testUnsubAll()
#------------------------
def testSend():
publisher = Publisher()
called = []
class TestListener:
def __init__(self, num):
self.number = num
def __call__(self, b):
called.append( 'TL%scb' % self.number )
def notify(self, b):
called.append( 'TL%sm' % self.number )
def funcListener(b):
called.append('func')
lisnr1 = TestListener(1)
lisnr2 = TestListener(2)
lisnr3 = funcListener
lisnr4 = lambda x: called.append('lambda')
topic1 = 'politics'
topic2 = 'history'
topic3 = ('politics','UN')
topic4 = ('politics','NATO','US')
topic5 = ('politics','NATO')
publisher.subscribe(lisnr1, topic1)
publisher.subscribe(lisnr2, topic2)
publisher.subscribe(lisnr2.notify, topic2)
publisher.subscribe(lisnr3, topic4)
publisher.subscribe(lisnr4)
print publisher
# setup ok, now test send/receipt
publisher.sendMessage(topic1)
assert called == ['lambda','TL1cb']
called = []
publisher.sendMessage(topic2)
assert called == ['lambda','TL2cb','TL2m']
called = []
publisher.sendMessage(topic3)
assert called == ['lambda','TL1cb']
called = []
publisher.sendMessage(topic4)
assert called == ['lambda','TL1cb','func']
called = []
publisher.sendMessage(topic5)
assert called == ['lambda','TL1cb']
assert publisher.getDeliveryCount() == 12
assert publisher.getMessageCount() == 5
# test weak referencing works:
_NodeCallback.notified = 0
del lisnr2
called = []
publisher.sendMessage(topic2)
assert called == ['lambda']
assert _NodeCallback.notified == 2
done('testSend')
testSend()
assert _NodeCallback.notified == 5
def testDead():
# verify if weak references work as expected
print '------ Starting testDead ----------'
node = _TopicTreeNode('t1', None)
lisnr1 = SimpleListener(1)
lisnr2 = SimpleListener(2)
lisnr3 = SimpleListener(3)
lisnr4 = SimpleListener(4)
node.addCallable(lisnr1)
node.addCallable(lisnr2)
node.addCallable(lisnr3)
node.addCallable(lisnr4)
print 'Deleting listeners first'
_NodeCallback.notified = 0
del lisnr1
del lisnr2
assert _NodeCallback.notified == 2
print 'Deleting node first'
_NodeCallback.notified = 0
del node
del lisnr3
del lisnr4
assert _NodeCallback.notified == 0
lisnr1 = SimpleListener(1)
lisnr2 = SimpleListener(2)
lisnr3 = SimpleListener(3)
lisnr4 = SimpleListener(4)
# try same with root of tree
node = _TopicTreeRoot()
node.addTopic(('',), lisnr1)
node.addTopic(('',), lisnr2)
node.addTopic(('',), lisnr3)
node.addTopic(('',), lisnr4)
# add objects that will die immediately to see if cleanup occurs
# this must be done visually as it is a low-level detail
_NodeCallback.notified = 0
_TopicTreeRoot.callbackDeadLimit = 3
node.addTopic(('',), SimpleListener(5))
node.addTopic(('',), SimpleListener(6))
node.addTopic(('',), SimpleListener(7))
print node.numListeners()
assert node.numListeners() == (4, 3)
node.addTopic(('',), SimpleListener(8))
assert node.numListeners() == (4, 0)
assert _NodeCallback.notified == 4
print 'Deleting listeners first'
_NodeCallback.notified = 0
del lisnr1
del lisnr2
assert _NodeCallback.notified == 2
print 'Deleting node first'
_NodeCallback.notified = 0
del node
del lisnr3
del lisnr4
assert _NodeCallback.notified == 0
done('testDead')
testDead()
print 'Exiting tests'
#---------------------------------------------------------------------------
if __name__ == '__main__':
test()