-
Notifications
You must be signed in to change notification settings - Fork 32
/
handlers.py
76 lines (61 loc) · 2.57 KB
/
handlers.py
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
"""
handlers.py:
Defines a set of base classes that allow for the system to handle various functions of the system. Primarily this
defines the "DataHandler" base class for handling data.
@author mstarch
"""
import abc
class DataHandler(abc.ABC):
"""
Defines the necessary functions required to handle data as part of the F prime project. This allows any implementer
to be used to handle data.
"""
@abc.abstractmethod
def data_callback(self, data, sender=None):
"""
Callback function used to handle data being produced elsewhere in the system and processed by the given object.
Data supplied should be of a known type for the given object, and sender is an id of the sender. If not supplied
sender will be None.
:param data: data to be handled by this class
:param sender: (optional) id of sender, otherwise None
"""
class HandlerRegistrar(abc.ABC):
"""
Defines a class that will take in registrants and remember them for calling back later. These objects should be of
the type "DataHandler" as this handler will send data back to these handlers when asked to do so.
"""
def __init__(self):
"""
Constructor defining the internal lists needed to store the registrants.
"""
super().__init__()
self._registrants = []
def register(self, registrant):
"""
Register a registrant with this registrar. Will be stored and called back when asked to send data to all the
handlers registered.
:param registrant: handler to register
"""
if not isinstance(registrant, DataHandler):
raise ValueError("Cannot register non data handler")
self._registrants.append(registrant)
def deregister(self, registrant):
"""
Remove a registrant from the registrar such that it will not be called back later. Note: ignores invalid
removals by trapping the error, as the desired effect is already satisfied.
:param registrant: registrant to remove
:return: True if found, False if not. May safely be ignored.
"""
try:
self._registrants.remove(registrant)
return True
except ValueError:
return False
def send_to_all(self, data, sender=None):
"""
Sends the given data to all registrants.
:param data: data to send back to registrants
:param sender: (optional) sender to pass to data_callback
"""
for registrant in self._registrants:
registrant.data_callback(data, sender)