import { Observable } from 'rxjs/Observable' ; import { BehaviorSubject } from 'rxjs/BehaviorSubject' ; import { ReplaySubject } from 'rxjs/ReplaySubject' ; import { Subject } from 'rxjs/Subject' ; import 'rxjs/add/observable/timer' ; import 'rxjs/add/operator/do' ; import 'rxjs/add/operator/filter' ; import 'rxjs/add/operator/first' ; import 'rxjs/add/operator/map' ; import 'rxjs/add/operator/share' ; import 'rxjs/add/operator/switchMap' ; let _verbose = true ; export function getVerbose ( ) { return _verbose ; } export function setVerbose ( value ) { _verbose = value ; } /** Cached values expire after this period (ms) by default */ export const defaultExpirationPeriod = 3000 ; /** Caching event notification callback */ export type Notifier = ( msg : NotificationMessage ) => void ; param { Observable < T > } source param { Observable < any > } updateWhen param { T | ReplaySubject < T > | BehaviorSubject < T > } initialValue export function createCache < T > ( source , updateWhen , initialValue ? ) { const cacheSubject = getSubject < T > ( initialValue ) ; updateWhen . switchMap ( ( ) => source . first ( ) ) . subscribe ( data => cacheSubject . next ( data ) ) ; return cacheSubject . asObservable ( ) ; } ; export function createOnDemandCache < T > ( source , updateWhen , notifier ? , expirationPeriod ? , initialValue ? ) { let expired = 0 ; let firstTime = true ; if ( ! notifier ) { notifier = NullNotifier ; } expirationPeriod = ( expirationPeriod == null ) ? defaultExpirationPeriod : expirationPeriod ; return createCache ( source . do ( data => { expired = Date . now ( ) + expirationPeriod ; notifier ( new NotificationMessage ( 'fetched' , 'Fetched' , data ) ) ; } ) , updateWhen . map ( force => firstTime || force || expired < Date . now ( ) ) . filter ( fetch => { firstTime = false ; notifier ( fetch ? new NotificationMessage ( 'fetching' , 'Fetching...' ) : new NotificationMessage ( 'cached' , 'Use cached value.' ) ) ; return fetch ; } ) , initialValue ) ; } ; export function createTimerCache < T > ( source , expirationPeriod = defaultExpirationPeriod , initialValue ? ) { const cacheSubject = getSubject < T > ( initialValue ) ; return Observable . timer ( 0 , expirationPeriod ) . switchMap ( count => { log ( 'source executing #' + count ) ; const next = source . first ( ) . do ( data => log ( 'Fetched data arrived' ) ) . do ( data => cacheSubject . next ( data ) ) . switchMap ( ( ) => cacheSubject ) ; return count === 0 ? next . subscribe ( ) && cacheSubject : next ; } ) . share ( ) ; } /** Message provided to notifiers. */ export class NotificationMessage { constructor ( public type , public message ? , public body ? ) { body ? log ( message , body ) : log ( message ) ; } } export function NullNotifier ( msg ) { } export function log ( ... args ) { if ( _verbose ) { console . log . apply ( null , args ) ; } } function getSubject < T > ( value ? ) { return value instanceof Subject ? value : new BehaviorSubject ( value ) ; }