import { Subject , Observable , throwError , timer , Subscription } from 'rxjs' ; import { noop } from 'lodash' ; import { exhaustMap , tap , takeUntil , switchMap , switchMapTo , catchError } from 'rxjs/operators' ; const DEFAULT_HEARTBEAT_INTERVAL = 1000 ; const DEFAULT_INACTIVITY_TIMEOUT = 0 ; export interface ObservableMonitorOptions < E > { heartbeatInterval ? : number ; inactivityTimeout ? : number ; onError ? : ( err : E ) => void ; } export function createObservableMonitor < T , E > ( observableFactory , { heartbeatInterval = DEFAULT_HEARTBEAT_INTERVAL , inactivityTimeout = DEFAULT_INACTIVITY_TIMEOUT , onError = noop , } = { } ) { return new Observable ( ( subscriber ) => { const subscription = timer ( 0 , heartbeatInterval ) . pipe ( exhaustMap ( ( ) => takeUntilDurationOfInactivity ( observableFactory ( ) , inactivityTimeout ) ) , catchError ( ( err , source$ ) => { onError ( err ) ; return source$ ; } ) ) . subscribe ( subscriber ) ; return ( ) => { subscription . unsubscribe ( ) ; } ; } ) ; } function takeUntilDurationOfInactivity < T > ( source$ , inactivityTimeout ) { if ( inactivityTimeout ) { const inactivityMonitor$ = new Subject < void > ( ) ; return source$ . pipe ( takeUntil ( inactivityMonitor$ . pipe ( switchMap ( ( ) => timer ( inactivityTimeout ) ) , switchMapTo ( throwError ( new Error ( `Observable Monitor: Hung Observable restarted after ${ inactivityTimeout }ms of inactivity` ) ) ) ) ) , tap ( ( ) => inactivityMonitor$ . next ( ) ) ) ; } return source$ ; }