<s> from rx . observable import Observable <EOL> from rx . anonymousobservable import AnonymousObservable <EOL> from rx . internal . utils import is_future <EOL> from rx . internal import extensionclassmethod <EOL> @ extensionclassmethod ( Observable ) <EOL> def from_future ( cls , future ) : <EOL> """<STR_LIT>""" <EOL> def subscribe ( observer ) : <EOL> def done ( future ) : <EOL> try : <EOL> value = future . result ( ) <EOL> except Exception as ex : <EOL> observer . on_error ( ex ) <EOL> else : <EOL> observer . on_next ( value ) <EOL> observer . on_completed ( ) <EOL> future . add_done_callback ( done ) <EOL> def dispose ( ) : <EOL> if future and future . cancel : <EOL> future . cancel ( ) <EOL> return dispose <EOL> return AnonymousObservable ( subscribe ) if is_future ( future ) else future </s>
