<s> from rx . observable import Observable <EOL> from rx . anonymousobservable import AnonymousObservable <EOL> from rx . disposables import CompositeDisposable , SingleAssignmentDisposable <EOL> from rx . internal import extensionmethod , extensionclassmethod <EOL> @ extensionmethod ( Observable , instancemethod = True ) <EOL> def combine_latest ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> args = list ( args ) <EOL> if args and isinstance ( args [ <NUM_LIT:0> ] , list ) : <EOL> args = args [ <NUM_LIT:0> ] <EOL> args . insert ( <NUM_LIT:0> , self ) <EOL> return Observable . combine_latest ( * args ) <EOL> @ extensionclassmethod ( Observable ) <EOL> def combine_latest ( cls , * args ) : <EOL> """<STR_LIT>""" <EOL> args = list ( args ) <EOL> result_selector = args . pop ( ) <EOL> if isinstance ( args [ <NUM_LIT:0> ] , list ) : <EOL> args = args [ <NUM_LIT:0> ] <EOL> parent = args [ <NUM_LIT:0> ] <EOL> def subscribe ( observer ) : <EOL> n = len ( args ) <EOL> has_value = [ False ] * n <EOL> has_value_all = [ False ] <EOL> is_done = [ False ] * n <EOL> values = [ None ] * n <EOL> def next ( i ) : <EOL> has_value [ i ] = True <EOL> if has_value_all [ <NUM_LIT:0> ] or all ( has_value ) : <EOL> try : <EOL> res = result_selector ( * values ) <EOL> except Exception as ex : <EOL> observer . on_error ( ex ) <EOL> return <EOL> observer . on_next ( res ) <EOL> elif all ( [ x for j , x in enumerate ( is_done ) if j != i ] ) : <EOL> observer . on_completed ( ) <EOL> has_value_all [ <NUM_LIT:0> ] = all ( has_value ) <EOL> def done ( i ) : <EOL> is_done [ i ] = True <EOL> if all ( is_done ) : <EOL> observer . on_completed ( ) <EOL> subscriptions = [ None ] * n <EOL> def func ( i ) : <EOL> subscriptions [ i ] = SingleAssignmentDisposable ( ) <EOL> def on_next ( x ) : <EOL> with parent . lock : <EOL> values [ i ] = x <EOL> next ( i ) <EOL> def on_completed ( ) : <EOL> with parent . lock : <EOL> done ( i ) <EOL> subscriptions [ i ] . disposable = args [ i ] . subscribe ( on_next , observer . on_error , on_completed ) <EOL> for idx in range ( n ) : <EOL> func ( idx ) <EOL> return CompositeDisposable ( subscriptions ) <EOL> return AnonymousObservable ( subscribe ) </s>
