/usr/lib/python3/dist-packages/wormhole/_lister.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 69 70 71 72 73 | from __future__ import print_function, absolute_import, unicode_literals
from zope.interface import implementer
from attr import attrs, attrib
from attr.validators import provides
from automat import MethodicalMachine
from . import _interfaces
@attrs
@implementer(_interfaces.ILister)
class Lister(object):
_timing = attrib(validator=provides(_interfaces.ITiming))
m = MethodicalMachine()
set_trace = getattr(m, "_setTrace", lambda self, f: None)
def wire(self, rendezvous_connector, input):
self._RC = _interfaces.IRendezvousConnector(rendezvous_connector)
self._I = _interfaces.IInput(input)
# Ideally, each API request would spawn a new "list_nameplates" message
# to the server, so the response would be maximally fresh, but that would
# require correlating server request+response messages, and the protocol
# is intended to be less stateful than that. So we offer a weaker
# freshness property: if no server requests are in flight, then a new API
# request will provoke a new server request, and the result will be
# fresh. But if a server request is already in flight when a second API
# request arrives, both requests will be satisfied by the same response.
@m.state(initial=True)
def S0A_idle_disconnected(self): pass # pragma: no cover
@m.state()
def S1A_wanting_disconnected(self): pass # pragma: no cover
@m.state()
def S0B_idle_connected(self): pass # pragma: no cover
@m.state()
def S1B_wanting_connected(self): pass # pragma: no cover
@m.input()
def connected(self): pass
@m.input()
def lost(self): pass
@m.input()
def refresh(self): pass
@m.input()
def rx_nameplates(self, all_nameplates): pass
@m.output()
def RC_tx_list(self):
self._RC.tx_list()
@m.output()
def I_got_nameplates(self, all_nameplates):
# We get a set of nameplate ids. There may be more attributes in the
# future: change RendezvousConnector._response_handle_nameplates to
# get them
self._I.got_nameplates(all_nameplates)
S0A_idle_disconnected.upon(connected, enter=S0B_idle_connected, outputs=[])
S0B_idle_connected.upon(lost, enter=S0A_idle_disconnected, outputs=[])
S0A_idle_disconnected.upon(refresh,
enter=S1A_wanting_disconnected, outputs=[])
S1A_wanting_disconnected.upon(refresh,
enter=S1A_wanting_disconnected, outputs=[])
S1A_wanting_disconnected.upon(connected, enter=S1B_wanting_connected,
outputs=[RC_tx_list])
S0B_idle_connected.upon(refresh, enter=S1B_wanting_connected,
outputs=[RC_tx_list])
S0B_idle_connected.upon(rx_nameplates, enter=S0B_idle_connected,
outputs=[I_got_nameplates])
S1B_wanting_connected.upon(lost, enter=S1A_wanting_disconnected, outputs=[])
S1B_wanting_connected.upon(refresh, enter=S1B_wanting_connected,
outputs=[RC_tx_list])
S1B_wanting_connected.upon(rx_nameplates, enter=S0B_idle_connected,
outputs=[I_got_nameplates])
|