vici: Move event (un-)registration to a helper method in Python library
Also make sure events are unregistered in case of exceptions in streamed_request().
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
Copyright (c) 2015 Björn Schuberg
|
||||
Copyright (c) 2015 Martin Willi
|
||||
Copyright (c) 2015 Tobias Brunner
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
|
||||
@@ -8,3 +8,6 @@ class SessionException(Exception):
|
||||
|
||||
class CommandException(Exception):
|
||||
"""Command result exception."""
|
||||
|
||||
class EventUnknownException(Exception):
|
||||
"""Event unknown exception."""
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import collections
|
||||
import socket
|
||||
|
||||
from .exception import SessionException, CommandException
|
||||
from .exception import SessionException, CommandException, EventUnknownException
|
||||
from .protocol import Transport, Packet, Message
|
||||
|
||||
|
||||
@@ -215,6 +215,32 @@ class SessionHandler(object):
|
||||
self.transport.send(packet)
|
||||
return Packet.parse(self.transport.receive())
|
||||
|
||||
def _register_unregister(self, event_type, register):
|
||||
"""Register or unregister for the given event.
|
||||
|
||||
:param event_type: event to register
|
||||
:type event_type: str
|
||||
:param register: whether to register or unregister
|
||||
:type register: bool
|
||||
"""
|
||||
if register:
|
||||
packet = Packet.register_event(event_type)
|
||||
else:
|
||||
packet = Packet.unregister_event(event_type)
|
||||
response = self._communicate(packet)
|
||||
if response.response_type == Packet.EVENT_UNKNOWN:
|
||||
raise EventUnknownException(
|
||||
"Unknown event type '{event}'".format(event=event_type)
|
||||
)
|
||||
elif response.response_type != Packet.EVENT_CONFIRM:
|
||||
raise SessionException(
|
||||
"Unexpected response type {type}, "
|
||||
"expected '{confirm}' (EVENT_CONFIRM)".format(
|
||||
type=response.response_type,
|
||||
confirm=Packet.EVENT_CONFIRM,
|
||||
)
|
||||
)
|
||||
|
||||
def request(self, command, message=None):
|
||||
"""Send request with an optional message.
|
||||
|
||||
@@ -265,57 +291,37 @@ class SessionHandler(object):
|
||||
if message is not None:
|
||||
message = Message.serialize(message)
|
||||
|
||||
# subscribe to event stream
|
||||
packet = Packet.register_event(event_stream_type)
|
||||
response = self._communicate(packet)
|
||||
self._register_unregister(event_stream_type, True);
|
||||
|
||||
if response.response_type != Packet.EVENT_CONFIRM:
|
||||
raise SessionException(
|
||||
"Unexpected response type {type}, "
|
||||
"expected '{confirm}' (EVENT_CONFIRM)".format(
|
||||
type=response.response_type,
|
||||
confirm=Packet.EVENT_CONFIRM,
|
||||
)
|
||||
)
|
||||
try:
|
||||
packet = Packet.request(command, message)
|
||||
self.transport.send(packet)
|
||||
exited = False
|
||||
while True:
|
||||
response = Packet.parse(self.transport.receive())
|
||||
if response.response_type == Packet.EVENT:
|
||||
if not exited:
|
||||
try:
|
||||
yield Message.deserialize(response.payload)
|
||||
except GeneratorExit:
|
||||
exited = True
|
||||
pass
|
||||
else:
|
||||
break
|
||||
|
||||
# issue command, and read any event messages
|
||||
packet = Packet.request(command, message)
|
||||
self.transport.send(packet)
|
||||
exited = False
|
||||
while True:
|
||||
response = Packet.parse(self.transport.receive())
|
||||
if response.response_type == Packet.EVENT:
|
||||
if not exited:
|
||||
try:
|
||||
yield Message.deserialize(response.payload)
|
||||
except GeneratorExit:
|
||||
exited = True
|
||||
pass
|
||||
if response.response_type == Packet.CMD_RESPONSE:
|
||||
command_response = Message.deserialize(response.payload)
|
||||
else:
|
||||
break
|
||||
|
||||
if response.response_type == Packet.CMD_RESPONSE:
|
||||
command_response = Message.deserialize(response.payload)
|
||||
else:
|
||||
raise SessionException(
|
||||
"Unexpected response type {type}, "
|
||||
"expected '{response}' (CMD_RESPONSE)".format(
|
||||
type=response.response_type,
|
||||
response=Packet.CMD_RESPONSE
|
||||
raise SessionException(
|
||||
"Unexpected response type {type}, "
|
||||
"expected '{response}' (CMD_RESPONSE)".format(
|
||||
type=response.response_type,
|
||||
response=Packet.CMD_RESPONSE
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
# unsubscribe from event stream
|
||||
packet = Packet.unregister_event(event_stream_type)
|
||||
response = self._communicate(packet)
|
||||
if response.response_type != Packet.EVENT_CONFIRM:
|
||||
raise SessionException(
|
||||
"Unexpected response type {type}, "
|
||||
"expected '{confirm}' (EVENT_CONFIRM)".format(
|
||||
type=response.response_type,
|
||||
confirm=Packet.EVENT_CONFIRM,
|
||||
)
|
||||
)
|
||||
finally:
|
||||
self._register_unregister(event_stream_type, False);
|
||||
|
||||
# evaluate command result, if any
|
||||
if "success" in command_response:
|
||||
|
||||
Reference in New Issue
Block a user