Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
waiting_s.add_transition(Transition(waiting_r, lambda x: x[:2] != 's ', die('Expected "s " while parsing routers not "%s"')))
waiting_s.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', nothing))
waiting_w.add_transition(Transition(waiting_p, lambda x: x[:2] == 'w ', self._router_bandwidth))
waiting_w.add_transition(Transition(waiting_r, ignorable_line, nothing))
waiting_w.add_transition(Transition(waiting_s, lambda x: x[:2] == 'r ', self._router_begin)) # "w" lines are optional
waiting_w.add_transition(Transition(waiting_r, lambda x: x[:2] != 'w ', die('Expected "w " while parsing routers not "%s"')))
waiting_w.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', nothing))
waiting_p.add_transition(Transition(waiting_r, lambda x: x[:2] == 'p ', self._router_policy))
waiting_p.add_transition(Transition(waiting_r, ignorable_line, nothing))
waiting_p.add_transition(Transition(waiting_s, lambda x: x[:2] == 'r ', self._router_begin)) # "p" lines are optional
waiting_p.add_transition(Transition(waiting_r, lambda x: x[:2] != 'p ', die('Expected "p " while parsing routers not "%s"')))
waiting_p.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', nothing))
self._network_status_parser = FSM([waiting_r, waiting_s, waiting_w, waiting_p])
if write_state_diagram:
with open('routerfsm.dot', 'w') as fsmfile:
fsmfile.write(self._network_status_parser.dotty())
self.post_bootstrap = defer.Deferred()
if bootstrap:
if self.protocol.post_bootstrap:
self.protocol.post_bootstrap.addCallback(self._bootstrap).addErrback(self.post_bootstrap.errback)
else:
self._bootstrap()
waiting_s.add_transition(Transition(waiting_w, lambda x: x[:2] == 's ', self._router_flags))
waiting_s.add_transition(Transition(eat_line, lambda x: x[:2] != 's ', die('Expected "s " while parsing routers not "%s"')))
waiting_s.add_transition(Transition(eat_line, lambda x: x.strip() == '.', nothing))
waiting_w.add_transition(Transition(waiting_p, lambda x: x[:2] == 'w ', self._router_bandwidth))
waiting_w.add_transition(Transition(waiting_s, lambda x: x[:2] == 'r ', self._router_begin)) # "w" lines are optional
waiting_w.add_transition(Transition(eat_line, lambda x: x[:2] != 'w ', die('Expected "w " while parsing routers not "%s"')))
waiting_w.add_transition(Transition(eat_line, lambda x: x.strip() == '.', nothing))
waiting_p.add_transition(Transition(waiting_r, lambda x: x[:2] == 'p ', self._router_policy))
waiting_p.add_transition(Transition(waiting_s, lambda x: x[:2] == 'r ', self._router_begin)) # "p" lines are optional
waiting_p.add_transition(Transition(eat_line, lambda x: x[:2] != 'p ', die('Expected "p " while parsing routers not "%s"')))
waiting_p.add_transition(Transition(eat_line, lambda x: x.strip() == '.', nothing))
self._network_status_parser = FSM([eat_line, waiting_r, waiting_s, waiting_w, waiting_p])
self.post_bootstrap = defer.Deferred()
if bootstrap:
if self.protocol.post_bootstrap:
self.protocol.post_bootstrap.addCallback(self._bootstrap).addErrback(self.post_bootstrap.errback)
else:
self._bootstrap()
recv.add_transition(Transition(recv,
self._is_continuation_line,
self._accumulate_response))
recv.add_transition(Transition(idle,
self._is_finish_line,
self._broadcast_response))
recvmulti.add_transition(Transition(recv,
self._is_end_line,
lambda x: None))
recvmulti.add_transition(Transition(recvmulti,
self._is_not_end_line,
self._accumulate_multi_response))
self.fsm = FSM([recvnotify, idle, recvmulti, recv])
self.state_idle = idle
## hand-set initial state default start state is first in the
## list; the above looks nice in dotty though
self.fsm.state = idle
if DEBUG:
self.debuglog = open('txtorcon-debug.log', 'w')
with open('fsm.dot', 'w') as fsmfile:
fsmfile.write(self.fsm.dotty())