#NEWLINE# @ extensionmethod ( Observable ) #NEWLINE# def group_join( self , right , left_duration_selector , right_duration_selector , result_selector ) : #NEWLINE# #INDENT# "Correlates the elements of two sequences based on overlapping\n    durations, and groups the results.\n\n    Keyword arguments:\n    right -- The right observable sequence to join elements for.\n    left_duration_selector -- A function to select the duration (expressed\n        as an observable sequence) of each element of the left observable\n        sequence, used to determine overlap.\n    right_duration_selector -- A function to select the duration (expressed\n        as an observable sequence) of each element of the right observable\n        sequence, used to determine overlap.\n    result_selector -- A function invoked to compute a result element for\n        any element of the left sequence with overlapping elements from the\n        right observable sequence. The first parameter passed to the\n        function is an element of the left sequence. The second parameter\n        passed to the function is an observable sequence with elements from\n        the right sequence that overlap with the left sequence's element.\n\n    Returns an observable sequence that contains result elements computed\n    from source elements that have an overlapping duration.\n    " #NEWLINE# left = self #NEWLINE# def subscribe( observer ) : #NEWLINE# #INDENT# nothing = ( lambda _ : None ) #NEWLINE# group = CompositeDisposable ( ) #NEWLINE# r = RefCountDisposable ( group ) #NEWLINE# left_map = OrderedDict ( ) #NEWLINE# right_map = OrderedDict ( ) #NEWLINE# left_id = [ 0 ] #NEWLINE# right_id = [ 0 ] #NEWLINE# def on_next_left( value ) : #NEWLINE# #INDENT# s = Subject ( ) #NEWLINE# with self . lock : #NEWLINE# #INDENT# _id = left_id [ 0 ] #NEWLINE# left_id [ 0 ] += 1 #NEWLINE# left_map [ _id ] = s #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# result = result_selector ( value , add_ref ( s , r ) ) #NEWLINE# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# log . error ( ( '*** Exception: %s' % e ) ) #NEWLINE# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# return #NEWLINE# #UNINDENT# observer . on_next ( result ) #NEWLINE# for right_value in right_map . values ( ) : #NEWLINE# #INDENT# s . on_next ( right_value ) #NEWLINE# #UNINDENT# md = SingleAssignmentDisposable ( ) #NEWLINE# group . add ( md ) #NEWLINE# def expire( ) : #NEWLINE# #INDENT# if ( _id in left_map ) : #NEWLINE# #INDENT# del left_map [ _id ] #NEWLINE# s . on_completed ( ) #NEWLINE# #UNINDENT# group . remove ( md ) #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# duration = left_duration_selector ( value ) #NEWLINE# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# return #NEWLINE# #UNINDENT# def on_error( e ) : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# #UNINDENT# md . disposable = duration . take ( 1 ) . subscribe ( nothing , on_error , expire ) #NEWLINE# #UNINDENT# def on_error_left( e ) : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# #UNINDENT# group . add ( left . subscribe ( on_next_left , on_error_left , observer . on_completed ) ) #NEWLINE# def on_next_right( value ) : #NEWLINE# #INDENT# with self . lock : #NEWLINE# #INDENT# _id = right_id [ 0 ] #NEWLINE# right_id [ 0 ] += 1 #NEWLINE# right_map [ _id ] = value #NEWLINE# #UNINDENT# md = SingleAssignmentDisposable ( ) #NEWLINE# group . add ( md ) #NEWLINE# def expire( ) : #NEWLINE# #INDENT# del right_map [ _id ] #NEWLINE# group . remove ( md ) #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# duration = right_duration_selector ( value ) #NEWLINE# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# return #NEWLINE# #UNINDENT# def on_error( e ) : #NEWLINE# #INDENT# with self . lock : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# #UNINDENT# #UNINDENT# md . disposable = duration . take ( 1 ) . subscribe ( nothing , on_error , expire ) #NEWLINE# with self . lock : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_next ( value ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# def on_error_right( e ) : #NEWLINE# #INDENT# for left_value in left_map . values ( ) : #NEWLINE# #INDENT# left_value . on_error ( e ) #NEWLINE# #UNINDENT# observer . on_error ( e ) #NEWLINE# #UNINDENT# group . add ( right . subscribe ( on_next_right , on_error_right ) ) #NEWLINE# return r #NEWLINE# #UNINDENT# return AnonymousObservable ( right )