#NEWLINE# def observable_delay_timespan( source , duetime , scheduler ) : #NEWLINE# #INDENT# duetime = scheduler . to_timedelta ( subscribe ) #NEWLINE# def subscribe( observer ) : #NEWLINE# #INDENT# cancelable = SerialDisposable ( ) #NEWLINE# exception = [ None ] #NEWLINE# active = [ False ] #NEWLINE# running = [ False ] #NEWLINE# queue = [ ] #NEWLINE# def on_next( notification ) : #NEWLINE# #INDENT# log . debug ( 'observable_delay_timespan:subscribe:on_next()' ) #NEWLINE# should_run = False #NEWLINE# with source . lock : #NEWLINE# #INDENT# if ( notification . value . kind == 'E' ) : #NEWLINE# #INDENT# del queue [ : ] #NEWLINE# queue . append ( notification ) #NEWLINE# exception [ 0 ] = notification . value . exception #NEWLINE# should_run = ( not running [ 0 ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# queue . append ( Timestamp ( value = notification . value , timestamp = ( notification . timestamp + duetime ) ) ) #NEWLINE# should_run = ( not active [ 0 ] ) #NEWLINE# active [ 0 ] = True #NEWLINE# #UNINDENT# #UNINDENT# if should_run : #NEWLINE# #INDENT# if exception [ 0 ] : #NEWLINE# #INDENT# log . error ( '*** Exception: %s' , exception [ 0 ] ) #NEWLINE# observer . on_error ( exception [ 0 ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# d = SingleAssignmentDisposable ( ) #NEWLINE# cancelable . disposable = d #NEWLINE# def action( this ) : #NEWLINE# #INDENT# if exception [ 0 ] : #NEWLINE# #INDENT# log . error ( 'observable_delay_timespan:subscribe:on_next:action(), exception: %s' , exception [ 0 ] ) #NEWLINE# return #NEWLINE# #UNINDENT# with source . lock : #NEWLINE# #INDENT# running [ 0 ] = True #NEWLINE# while True : #NEWLINE# #INDENT# result = None #NEWLINE# if ( len ( queue ) and ( queue [ 0 ] . timestamp <= scheduler . now ( ) ) ) : #NEWLINE# #INDENT# result = queue . pop ( 0 ) . value #NEWLINE# #UNINDENT# if result : #NEWLINE# #INDENT# result . accept ( observer ) #NEWLINE# #UNINDENT# if ( not result ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# should_recurse = False #NEWLINE# recurse_duetime = 0 #NEWLINE# if len ( queue ) : #NEWLINE# #INDENT# should_recurse = True #NEWLINE# diff = ( queue [ 0 ] . timestamp - scheduler . now ( ) ) #NEWLINE# zero = ( timedelta ( 0 ) if isinstance ( diff , timedelta ) else 0 ) #NEWLINE# recurse_duetime = max ( zero , diff ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# active [ 0 ] = False #NEWLINE# #UNINDENT# ex = exception [ 0 ] #NEWLINE# running [ 0 ] = False #NEWLINE# #UNINDENT# if ex : #NEWLINE# #INDENT# observer . on_error ( ex ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if should_recurse : #NEWLINE# #INDENT# this ( recurse_duetime ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# d . disposable = scheduler . schedule_recursive_with_relative ( duetime , action ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# subscription = source . materialize ( ) . timestamp ( scheduler ) . subscribe ( on_next ) #NEWLINE# return CompositeDisposable ( subscription , cancelable ) #NEWLINE# #UNINDENT# return AnonymousObservable ( subscribe )