improve reply semantics

This commit is contained in:
Maximilian Hils 2016-08-08 19:20:06 -07:00
parent a52a1df23c
commit 0ee1b40c17

View File

@ -207,32 +207,29 @@ def handler(f):
# acking is ours. If not, it's someone else's and we ignore it. # acking is ours. If not, it's someone else's and we ignore it.
handling = False handling = False
# We're the first handler - ack responsibility is ours # We're the first handler - ack responsibility is ours
if not message.reply.handled: if message.reply.state == "unhandled":
handling = True handling = True
message.reply.handled = True message.reply.handle()
with master.handlecontext(): with master.handlecontext():
ret = f(master, message) ret = f(master, message)
if handling: if handling:
master.addons(f.__name__, message) master.addons(f.__name__, message)
if handling and not message.reply.acked and not message.reply.taken:
message.reply.ack()
# Reset the handled flag - it's common for us to feed the same object # Reset the handled flag - it's common for us to feed the same object
# through handlers repeatedly, so we don't want this to persist across # through handlers repeatedly, so we don't want this to persist across
# calls. # calls.
if handling: if handling and not message.reply.taken:
if not message.reply.taken: if message.reply.value == NO_REPLY:
message.reply.commit() message.reply.ack()
message.reply.handled = False message.reply.commit()
return ret return ret
# Mark this function as a handler wrapper # Mark this function as a handler wrapper
wrapper.__dict__["__handler"] = True wrapper.__dict__["__handler"] = True
return wrapper return wrapper
NO_REPLY = object() NO_REPLY = object() # special object we can distinguish from a valid "None" reply.
class Reply(object): class Reply(object):
@ -244,14 +241,37 @@ class Reply(object):
def __init__(self, obj): def __init__(self, obj):
self.obj = obj self.obj = obj
self.q = queue.Queue() self.q = queue.Queue()
# Has this message been acked?
self.acked = False self.state = "unhandled" # "unhandled" -> "handled" -> "taken" -> "committed"
# What's the ack message? self.value = NO_REPLY # holds the reply value. May change before things are actually commited.
self.ack_message = NO_REPLY
# Has the user taken responsibility for ack-ing? def handle(self):
self.taken = False """
# Has a handler taken responsibility for ack-ing? Reply are handled by controller.handlers, which may be nested. The first handler takes
self.handled = False responsibility and handles the reply.
"""
if self.state != "unhandled":
raise exceptions.ControlException("Message is {}, but expected it to be unhandled.".format(self.state))
self.state = "handled"
def take(self):
"""
Scripts or other parties make "take" a reply out of a normal flow.
For example, intercepted flows are taken out so that the connection thread does not proceed.
"""
if self.state != "handled":
raise exceptions.ControlException("Message is {}, but expected it to be handled.".format(self.state))
self.state = "taken"
def commit(self):
"""
Ultimately, messages are commited. This is done either automatically by the handler
if the message is not taken or manually by the entity which called .take().
"""
if self.state != "taken":
raise exceptions.ControlException("Message is {}, but expected it to be taken.".format(self.state))
self.state = "committed"
self.q.put(self.value)
def ack(self): def ack(self):
self.send(self.obj) self.send(self.obj)
@ -259,52 +279,32 @@ class Reply(object):
def kill(self): def kill(self):
self.send(exceptions.Kill) self.send(exceptions.Kill)
def take(self):
self.taken = True
def send(self, msg): def send(self, msg):
if self.acked: if self.state not in ("handled", "taken"):
raise exceptions.ControlException("Message already acked.") raise exceptions.ControlException(
if self.ack_message != NO_REPLY: "Reply is currently {}, did not expect a call to .send().".format(self.state)
# We may have overrides for this later. )
raise exceptions.ControlException("Message already has a response.") if self.value is not NO_REPLY:
self.acked = True raise exceptions.ControlException("There is already a reply for this message.")
self.ack_message = msg self.value = msg
if self.taken:
self.commit()
def commit(self):
"""
This is called by the handler to actually send the ack message.
"""
if self.ack_message == NO_REPLY:
raise exceptions.ControlException("Message has no response.")
self.q.put(self.ack_message)
def __del__(self): def __del__(self):
if not self.acked: if self.state != "comitted":
# This will be ignored by the interpreter, but emit a warning # This will be ignored by the interpreter, but emit a warning
raise exceptions.ControlException("Un-acked message: %s" % self.obj) raise exceptions.ControlException("Uncomitted message: %s" % self.obj)
class DummyReply(object): class DummyReply(Reply):
""" """
A reply object that does nothing. Useful when we need an object to seem A reply object that is not connected to anything. In contrast to regular Reply objects,
like it has a channel, and during testing. DummyReply objects are reset to "unhandled" after a commit so that they can be used
multiple times. Useful when we need an object to seem like it has a channel,
and during testing.
""" """
def __init__(self): def __init__(self):
self.acked = False super(DummyReply, self).__init__(None)
self.taken = False
self.handled = False
def kill(self): def commit(self):
self.send(None) super(DummyReply, self).commit()
self.state = "unhandled"
def ack(self): self.value = NO_REPLY
self.send(None)
def take(self):
self.taken = True
def send(self, msg):
self.acked = True