import { ConnectableObservable , Observable , Subject , from , merge } from 'rxjs' ; import { filter , map , pairwise , exhaustMap , publishReplay , share , take , takeUntil , } from 'rxjs/operators' ; import { hasLicenseInfoChanged } from './has_license_info_changed' ; import { ILicense } from './types' ; export function createLicenseUpdate ( triggerRefresh$ , stop$ , fetcher , initialValues ? ) { const manuallyRefresh$ = new Subject ( ) ; const fetched$ = merge ( triggerRefresh$ , manuallyRefresh$ ) . pipe ( exhaustMap ( fetcher ) , share ( ) ) ; const cached$ = fetched$ . pipe ( takeUntil ( stop$ ) , publishReplay ( 1 ) ) as ConnectableObservable < ILicense > ; const cachedSubscription = cached$ . connect ( ) ; stop$ . subscribe ( { complete : ( ) => cachedSubscription . unsubscribe ( ) } ) ; const initialValues$ = initialValues ? from ( [ undefined , initialValues ] ) : from ( [ undefined ] ) ; const license$ = merge ( initialValues$ , cached$ ) . pipe ( pairwise ( ) , filter ( ( [ previous , next ] ) => hasLicenseInfoChanged ( previous , next ! ) ) , map ( ( [ , next ] ) => next ! ) ) ; return { license$ , refreshManually ( ) { const licensePromise = fetched$ . pipe ( take ( 1 ) ) . toPromise ( ) ; manuallyRefresh$ . next ( ) ; return licensePromise ; } , } ; }