<s> import threading <EOL> from rx . blockingobservable import BlockingObservable <EOL> from rx . internal import extensionmethod <EOL> from rx . internal . enumerator import Enumerator <EOL> @ extensionmethod ( BlockingObservable ) <EOL> def to_iterable ( self ) : <EOL> """<STR_LIT>""" <EOL> condition = threading . Condition ( ) <EOL> notifications = [ ] <EOL> def on_next ( value ) : <EOL> """<STR_LIT>""" <EOL> condition . acquire ( ) <EOL> notifications . append ( value ) <EOL> condition . notify ( ) <EOL> condition . release ( ) <EOL> self . observable . materialize ( ) . subscribe ( on_next ) <EOL> def gen ( ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> condition . acquire ( ) <EOL> while not len ( notifications ) : <EOL> condition . wait ( ) <EOL> notification = notifications . pop ( <NUM_LIT:0> ) <EOL> if notification . kind == "<STR_LIT:E>" : <EOL> raise notification . exception <EOL> if notification . kind == "<STR_LIT:C>" : <EOL> return <EOL> condition . release ( ) <EOL> yield notification . value <EOL> return Enumerator ( gen ( ) ) <EOL> @ extensionmethod ( BlockingObservable ) <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . to_iterable ( ) </s>
