github/SleekXMPP
1
0
Fork 0
mirror of https://github.com/correl/SleekXMPP.git synced 2024-12-19 19:14:15 +00:00
Code Issues Projects Releases Packages Wiki Activity

context manager now returns a boolean 'result' as the context variable to indicate whether the transition timed out or if you are actually locked when entering the context body

Browse source
This commit is contained in:
Thom Nichols 2010-06-07 13:43:37 -04:00
parent 66cf0c2021
commit 47f1fb1690
2 changed files with 11 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
sleekxmpp/xmlstream/statemachine.py
View file

@ -182,12 +182,12 @@ class _StateCtx:
if time.time() >= start + self.wait: if time.time() >= start + self.wait:
logging.debug('StateMachine timeout while waiting for state: %s', self.from_state ) logging.debug('StateMachine timeout while waiting for state: %s', self.from_state )
self._timeout = True # to indicate we should not transition self._timeout = True # to indicate we should not transition
break return False
self.state_machine.lock.wait(self.wait) self.state_machine.lock.wait(self.wait)
logging.debug('StateMachine entered context in state: %s', logging.debug('StateMachine entered context in state: %s',
self.state_machine.current_state() ) self.state_machine.current_state() )
return self.state_machine return True
def __exit__(self, exc_type, exc_val, exc_tb): def __exit__(self, exc_type, exc_val, exc_tb):
if exc_val is not None: if exc_val is not None:

15
tests/test_statemachine.py
View file

@ -223,11 +223,12 @@ class testStateMachine(unittest.TestCase):
s = sm.StateMachine(('one','two','three')) s = sm.StateMachine(('one','two','three'))
with s.transition_ctx('two','three') as _s: with s.transition_ctx('two','three') as result:
self.assertTrue( _s['one'] ) self.failIf( result )
self.failIf( _s.current_state in ('two','three') ) self.assertTrue( s['one'] )
self.failIf( s.current_state in ('two','three') )
self.assertTrue( _s['one'] ) self.assertTrue( s['one'] )
def r1(): def r1():
print 'thread 1 started' print 'thread 1 started'
@ -237,10 +238,12 @@ class testStateMachine(unittest.TestCase):
def r2(): def r2():
print 'thread 2 started' print 'thread 2 started'
self.failIf( s['two'] ) self.failIf( s['two'] )
with s.transition_ctx('two','three', 10) as _s: with s.transition_ctx('two','three', 10) as result:
self.assertTrue( _s['two'] ) self.assertTrue( result )
self.assertTrue( s['two'] )
print 'thread 2 will transition on exit from the context manager...' print 'thread 2 will transition on exit from the context manager...'
self.assertTrue( s['three'] ) self.assertTrue( s['three'] )
print 'transitioned to %s' % s.current_state()
t1 = threading.Thread(target=r1) t1 = threading.Thread(target=r1)
t2 = threading.Thread(target=r2) t2 = threading.Thread(target=r2)