<s> from rx . observable import Observable <EOL> from rx . concurrency import timeout_scheduler <EOL> from rx . subjects import AsyncSubject <EOL> from rx . internal import extensionclassmethod <EOL> @ extensionclassmethod ( Observable ) <EOL> def to_async ( cls , func , scheduler = None ) : <EOL> """<STR_LIT>""" <EOL> scheduler = scheduler or timeout_scheduler <EOL> def wrapper ( * args ) : <EOL> subject = AsyncSubject ( ) <EOL> def action ( scheduler , state ) : <EOL> try : <EOL> result = func ( * args ) <EOL> except Exception as ex : <EOL> subject . on_error ( ex ) <EOL> return <EOL> subject . on_next ( result ) <EOL> subject . on_completed ( ) <EOL> scheduler . schedule ( action ) <EOL> return subject . as_observable ( ) <EOL> return wrapper </s>
