<s> import logging <EOL> from rx import AnonymousObservable , Observable <EOL> from rx . internal . utils import add_ref <EOL> from rx . internal import noop <EOL> from rx . disposables import SingleAssignmentDisposable , SerialDisposable , CompositeDisposable , RefCountDisposable <EOL> from rx . subjects import Subject <EOL> from rx . internal import extensionmethod <EOL> log = logging . getLogger ( "<STR_LIT>" ) <EOL> @ extensionmethod ( Observable ) <EOL> def window ( self , window_openings = None , window_closing_selector = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( window_openings , Observable ) and callable ( window_openings ) : <EOL> window_closing_selector = window_openings <EOL> window_openings = None <EOL> if window_openings and not window_closing_selector : <EOL> return observable_window_with_bounaries ( self , window_openings ) <EOL> if not window_openings and window_closing_selector : <EOL> return observable_window_with_closing_selector ( self , window_closing_selector ) <EOL> return observable_window_with_openings ( self , window_openings , window_closing_selector ) <EOL> def observable_window_with_openings ( self , window_openings , window_closing_selector ) : <EOL> return window_openings . group_join ( self , window_closing_selector , lambda _ : Observable . empty ( ) , lambda _ , window : window ) <EOL> def observable_window_with_bounaries ( self , window_boundaries ) : <EOL> source = self <EOL> def subscribe ( observer ) : <EOL> window = [ Subject ( ) ] <EOL> d = CompositeDisposable ( ) <EOL> r = RefCountDisposable ( d ) <EOL> observer . on_next ( add_ref ( window [ <NUM_LIT:0> ] , r ) ) <EOL> def on_next_window ( x ) : <EOL> window [ <NUM_LIT:0> ] . on_next ( x ) <EOL> def on_error ( err ) : <EOL> window [ <NUM_LIT:0> ] . on_error ( err ) <EOL> observer . on_error ( err ) <EOL> def on_completed ( ) : <EOL> window [ <NUM_LIT:0> ] . on_completed ( ) <EOL> observer . on_completed ( ) <EOL> d . add ( source . subscribe ( on_next_window , on_error , on_completed ) ) <EOL> def on_next_observer ( w ) : <EOL> window [ <NUM_LIT:0> ] . on_completed ( ) <EOL> window [ <NUM_LIT:0> ] = Subject ( ) <EOL> observer . on_next ( add_ref ( window [ <NUM_LIT:0> ] , r ) ) <EOL> d . add ( window_boundaries . subscribe ( on_next_observer , on_error , on_completed ) ) <EOL> return r <EOL> return AnonymousObservable ( subscribe ) <EOL> def observable_window_with_closing_selector ( self , window_closing_selector ) : <EOL> source = self <EOL> def subscribe ( observer ) : <EOL> m = SerialDisposable ( ) <EOL> d = CompositeDisposable ( m ) <EOL> r = RefCountDisposable ( d ) <EOL> window = [ Subject ( ) ] <EOL> observer . on_next ( add_ref ( window [ <NUM_LIT:0> ] , r ) ) <EOL> def on_next ( x ) : <EOL> window [ <NUM_LIT:0> ] . on_next ( x ) <EOL> def on_error ( ex ) : <EOL> window [ <NUM_LIT:0> ] . on_error ( ex ) <EOL> observer . on_error ( ex ) <EOL> def on_completed ( ) : <EOL> window [ <NUM_LIT:0> ] . on_completed ( ) <EOL> observer . on_completed ( ) <EOL> d . add ( source . subscribe ( on_next , on_error , on_completed ) ) <EOL> def create_window_close ( ) : <EOL> try : <EOL> window_close = window_closing_selector ( ) <EOL> except Exception as exception : <EOL> log . error ( "<STR_LIT>" % exception ) <EOL> observer . on_error ( exception ) <EOL> return <EOL> def on_completed ( ) : <EOL> window [ <NUM_LIT:0> ] . on_completed ( ) <EOL> window [ <NUM_LIT:0> ] = Subject ( ) <EOL> observer . on_next ( add_ref ( window [ <NUM_LIT:0> ] , r ) ) <EOL> create_window_close ( ) <EOL> m1 = SingleAssignmentDisposable ( ) <EOL> m . disposable = m1 <EOL> m1 . disposable = window_close . take ( <NUM_LIT:1> ) . subscribe ( noop , on_error , on_completed ) <EOL> create_window_close ( ) <EOL> return r <EOL> return AnonymousObservable ( subscribe ) </s>
