forked from hardbyte/python-can
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlistener.py
More file actions
64 lines (48 loc) · 1.65 KB
/
Copy pathlistener.py
File metadata and controls
64 lines (48 loc) · 1.65 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
try:
import queue
except ImportError:
import Queue as queue
class Listener(object):
def on_message_received(self, msg):
raise NotImplementedError(
"{} has not implemented on_message_received".format(
self.__class__.__name__)
)
def __call__(self, msg):
return self.on_message_received(msg)
def stop(self):
"""
Override to cleanup any open resources.
"""
class RedirectReader(Listener):
"""
A RedirectReader sends all received messages
to another Bus.
"""
def __init__(self, bus):
self.bus = bus
def on_message_received(self, msg):
self.bus.send(msg)
class BufferedReader(Listener):
"""
A BufferedReader is a subclass of :class:`~can.Listener` which implements a
**message buffer**: that is, when the :class:`can.BufferedReader` instance is
notified of a new message it pushes it into a queue of messages waiting to
be serviced.
"""
def __init__(self):
self.buffer = queue.Queue(0)
def on_message_received(self, msg):
self.buffer.put(msg)
def get_message(self, timeout=0.5):
"""
Attempts to retrieve the latest message received by the instance. If no message is
available it blocks for given timeout or until a message is received (whichever
is shorter),
:param float timeout: The number of seconds to wait for a new message.
:return: the :class:`~can.Message` if there is one, or None if there is not.
"""
try:
return self.buffer.get(block=True, timeout=timeout)
except queue.Empty:
return None