import * as Rx from 'rxjs' ; import { takeUntil } from 'rxjs/operators' ; param bufferSize export function shareWeakReplay < T > ( bufferSize ? ) { return ( source ) => { let subject ; const stop$ = new Rx . Subject ( ) ; return new Rx . Observable ( ( observer ) => { if ( ! subject ) { subject = new Rx . ReplaySubject < T > ( bufferSize ) ; } subject . subscribe ( observer ) . add ( ( ) => { if ( ! subject ) { return ; } if ( subject . observers . length === 0 ) { stop$ . next ( ) ; } if ( subject . closed || subject . isStopped ) { subject = undefined ; } } ) ; if ( subject && subject . observers . length === 1 ) { source . pipe ( takeUntil ( stop$ ) ) . subscribe ( subject ) ; } } ) ; } ; }