<s> from datetime import datetime <EOL> from rx . observable import Observable <EOL> from rx . anonymousobservable import AnonymousObservable <EOL> from rx . disposables import CompositeDisposable , SingleAssignmentDisposable , SerialDisposable <EOL> from rx . concurrency import timeout_scheduler <EOL> from rx . internal import extensionmethod <EOL> @ extensionmethod ( Observable ) <EOL> def timeout ( self , duetime , other = None , scheduler = None ) : <EOL> """<STR_LIT>""" <EOL> scheduler_method = None <EOL> source = self <EOL> other = other or Observable . throw_exception ( Exception ( "<STR_LIT>" ) ) <EOL> other = Observable . from_future ( other ) <EOL> scheduler = scheduler or timeout_scheduler <EOL> if isinstance ( duetime , datetime ) : <EOL> scheduler_method = scheduler . schedule_absolute <EOL> else : <EOL> scheduler_method = scheduler . schedule_relative <EOL> def subscribe ( observer ) : <EOL> switched = [ False ] <EOL> _id = [ <NUM_LIT:0> ] <EOL> original = SingleAssignmentDisposable ( ) <EOL> subscription = SerialDisposable ( ) <EOL> timer = SerialDisposable ( ) <EOL> subscription . disposable = original <EOL> def create_timer ( ) : <EOL> my_id = _id [ <NUM_LIT:0> ] <EOL> def action ( scheduler , state = None ) : <EOL> switched [ <NUM_LIT:0> ] = ( _id [ <NUM_LIT:0> ] == my_id ) <EOL> timer_wins = switched [ <NUM_LIT:0> ] <EOL> if timer_wins : <EOL> subscription . disposable = other . subscribe ( observer ) <EOL> timer . disposable = scheduler_method ( duetime , action ) <EOL> create_timer ( ) <EOL> def on_next ( x ) : <EOL> on_next_wins = not switched [ <NUM_LIT:0> ] <EOL> if on_next_wins : <EOL> _id [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> observer . on_next ( x ) <EOL> create_timer ( ) <EOL> def on_error ( e ) : <EOL> on_error_wins = not switched [ <NUM_LIT:0> ] <EOL> if on_error_wins : <EOL> _id [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> observer . on_error ( e ) <EOL> def on_completed ( ) : <EOL> on_completed_wins = not switched [ <NUM_LIT:0> ] <EOL> if on_completed_wins : <EOL> _id [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> observer . on_completed ( ) <EOL> original . disposable = source . subscribe ( on_next , on_error , on_completed ) <EOL> return CompositeDisposable ( subscription , timer ) <EOL> return AnonymousObservable ( subscribe ) </s>
