/usr/lib/python3/dist-packages/wormhole/_order.py is in magic-wormhole 0.10.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | from __future__ import print_function, absolute_import, unicode_literals
from zope.interface import implementer
from attr import attrs, attrib
from attr.validators import provides, instance_of
from automat import MethodicalMachine
from . import _interfaces
@attrs
@implementer(_interfaces.IOrder)
class Order(object):
_side = attrib(validator=instance_of(type(u"")))
_timing = attrib(validator=provides(_interfaces.ITiming))
m = MethodicalMachine()
set_trace = getattr(m, "_setTrace", lambda self, f: None)
def __attrs_post_init__(self):
self._key = None
self._queue = []
def wire(self, key, receive):
self._K = _interfaces.IKey(key)
self._R = _interfaces.IReceive(receive)
@m.state(initial=True)
def S0_no_pake(self): pass # pragma: no cover
@m.state(terminal=True)
def S1_yes_pake(self): pass # pragma: no cover
def got_message(self, side, phase, body):
#print("ORDER[%s].got_message(%s)" % (self._side, phase))
assert isinstance(side, type("")), type(phase)
assert isinstance(phase, type("")), type(phase)
assert isinstance(body, type(b"")), type(body)
if phase == "pake":
self.got_pake(side, phase, body)
else:
self.got_non_pake(side, phase, body)
@m.input()
def got_pake(self, side, phase, body): pass
@m.input()
def got_non_pake(self, side, phase, body): pass
@m.output()
def queue(self, side, phase, body):
assert isinstance(side, type("")), type(phase)
assert isinstance(phase, type("")), type(phase)
assert isinstance(body, type(b"")), type(body)
self._queue.append((side, phase, body))
@m.output()
def notify_key(self, side, phase, body):
self._K.got_pake(body)
@m.output()
def drain(self, side, phase, body):
del phase
del body
for (side, phase, body) in self._queue:
self._deliver(side, phase, body)
self._queue[:] = []
@m.output()
def deliver(self, side, phase, body):
self._deliver(side, phase, body)
def _deliver(self, side, phase, body):
self._R.got_message(side, phase, body)
S0_no_pake.upon(got_non_pake, enter=S0_no_pake, outputs=[queue])
S0_no_pake.upon(got_pake, enter=S1_yes_pake, outputs=[notify_key, drain])
S1_yes_pake.upon(got_non_pake, enter=S1_yes_pake, outputs=[deliver])
|