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
|
#!/usr/bin/env python
#
# Copyright 2008,2009 Free Software Foundation, Inc.
#
# This file is part of GNU Radio
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
#
"""
Abstract GNU Radio publisher/subscriber interface
This is a proof of concept implementation, will likely change significantly.
"""
class pubsub(dict):
def __init__(self):
self._publishers = {}
self._subscribers = {}
self._proxies = {}
def __missing__(self, key, value=None):
dict.__setitem__(self, key, value)
self._publishers[key] = None
self._subscribers[key] = []
self._proxies[key] = None
def __setitem__(self, key, val):
if key not in self:
self.__missing__(key, val)
elif self._proxies[key] is not None:
(p, newkey) = self._proxies[key]
p[newkey] = val
else:
dict.__setitem__(self, key, val)
for sub in self._subscribers[key]:
# Note this means subscribers will get called in the thread
# context of the 'set' caller.
sub(val)
def __getitem__(self, key):
if key not in self:
self.__missing__(key)
if self._proxies[key] is not None:
(p, newkey) = self._proxies[key]
return p[newkey]
elif self._publishers[key] is not None:
return self._publishers[key]()
else:
return dict.__getitem__(self, key)
def publish(self, key, publisher):
if key not in self:
self.__missing__(key)
if self._proxies[key] is not None:
(p, newkey) = self._proxies[key]
p.publish(newkey, publisher)
else:
self._publishers[key] = publisher
def subscribe(self, key, subscriber):
if key not in self:
self.__missing__(key)
if self._proxies[key] is not None:
(p, newkey) = self._proxies[key]
p.subscribe(newkey, subscriber)
else:
self._subscribers[key].append(subscriber)
def unpublish(self, key):
if self._proxies[key] is not None:
(p, newkey) = self._proxies[key]
p.unpublish(newkey)
else:
self._publishers[key] = None
def unsubscribe(self, key, subscriber):
if self._proxies[key] is not None:
(p, newkey) = self._proxies[key]
p.unsubscribe(newkey, subscriber)
else:
self._subscribers[key].remove(subscriber)
def proxy(self, key, p, newkey=None):
if key not in self:
self.__missing__(key)
if newkey is None:
newkey = key
self._proxies[key] = (p, newkey)
def unproxy(self, key):
self._proxies[key] = None
# Test code
if __name__ == "__main__":
import sys
o = pubsub()
# Non-existent key gets auto-created with None value
print("Auto-created key 'foo' value:", o['foo'])
# Add some subscribers
# First is a bare function
def print_len(x):
print("len=%i" % (len(x), ))
o.subscribe('foo', print_len)
# The second is a class member function
class subber(object):
def __init__(self, param):
self._param = param
def printer(self, x):
print(self._param, repr(x))
s = subber('param')
o.subscribe('foo', s.printer)
# The third is a lambda function
o.subscribe('foo', lambda x: sys.stdout.write('val=' + repr(x) + '\n'))
# Update key 'foo', will notify subscribers
print("Updating 'foo' with three subscribers:")
o['foo'] = 'bar'
# Remove first subscriber
o.unsubscribe('foo', print_len)
# Update now will only trigger second and third subscriber
print("Updating 'foo' after removing a subscriber:")
o['foo'] = 'bar2'
# Publish a key as a function, in this case, a lambda function
o.publish('baz', lambda: 42)
print("Published value of 'baz':", o['baz'])
# Unpublish the key
o.unpublish('baz')
# This will return None, as there is no publisher
print("Value of 'baz' with no publisher:", o['baz'])
# Set 'baz' key, it gets cached
o['baz'] = 'bazzz'
# Now will return cached value, since no provider
print("Cached value of 'baz' after being set:", o['baz'])
|
