vici: Allow the Python event listen() operation to optionally time out
The architecture of the Python client is completely blocking, which is fine for many simple scripts. For more complex applications that do other I/O and listen for vici events, the most feasible way to integrate the client is to use a dedicated thread. Unfortunately, Python has no simple support for thread cancellation. And having that thread in a blocking recv() does not allow to terminate the thread gracefully with an Event or the like. As a way out, add a timeout to the listen() call, so the thread can periodically do other things, like checking for termination Event and react on it. Returning from listen() on timeout can be suboptimal, though, as it involves registration/deregistration for events, including the risk for missing events while not registered. So return a (None, None) tuple instead on timeout, allowing the caller to periodically do other things while staying registered for the events and continue in listen(). The timeout applies to the socket recv() for the start of the header, only, so a message is either read in full or times out, avoiding the risk of breaking message framing on the stream with partial reads.
This commit is contained in:
committed by
Tobias Brunner
parent
11b18f65b1
commit
248a188d21
@@ -18,8 +18,8 @@ class Transport(object):
|
||||
def send(self, packet):
|
||||
self.socket.sendall(struct.pack("!I", len(packet)) + packet)
|
||||
|
||||
def receive(self):
|
||||
raw_length = self._recvall(self.HEADER_LENGTH)
|
||||
def receive(self, timeout=None):
|
||||
raw_length = self._recvall(self.HEADER_LENGTH, timeout)
|
||||
length, = struct.unpack("!I", raw_length)
|
||||
payload = self._recvall(length)
|
||||
return payload
|
||||
@@ -28,11 +28,14 @@ class Transport(object):
|
||||
self.socket.shutdown(socket.SHUT_RDWR)
|
||||
self.socket.close()
|
||||
|
||||
def _recvall(self, count):
|
||||
def _recvall(self, count, timeout=None):
|
||||
"""Ensure to read count bytes from the socket"""
|
||||
data = b""
|
||||
if count > 0:
|
||||
self.socket.settimeout(timeout)
|
||||
while len(data) < count:
|
||||
buf = self.socket.recv(count - len(data))
|
||||
self.socket.settimeout(None)
|
||||
if not buf:
|
||||
raise socket.error('Connection closed')
|
||||
data += buf
|
||||
|
||||
@@ -139,11 +139,19 @@ class Session(CommandWrappers, object):
|
||||
)
|
||||
)
|
||||
|
||||
def listen(self, event_types):
|
||||
def listen(self, event_types, timeout=None):
|
||||
"""Register and listen for the given events.
|
||||
|
||||
If a timeout is given, the generator produces a (None, None) tuple
|
||||
if no event has been received for that time. This allows the caller
|
||||
to either abort by breaking from the generator, or perform periodic
|
||||
tasks while staying registered within listen(), and then continue
|
||||
waiting for more events.
|
||||
|
||||
:param event_types: event types to register
|
||||
:type event_types: list
|
||||
:param timeout: timeout to wait for events, in fractions of a second
|
||||
:type timeout: float
|
||||
:return: generator for streamed event responses as (event_type, dict)
|
||||
:rtype: generator
|
||||
"""
|
||||
@@ -152,7 +160,11 @@ class Session(CommandWrappers, object):
|
||||
|
||||
try:
|
||||
while True:
|
||||
response = Packet.parse(self.transport.receive())
|
||||
try:
|
||||
response = Packet.parse(self.transport.receive(timeout))
|
||||
except socket.timeout:
|
||||
yield None, None
|
||||
continue
|
||||
if response.response_type == Packet.EVENT:
|
||||
try:
|
||||
msg = Message.deserialize(response.payload)
|
||||
|
||||
Reference in New Issue
Block a user