import { asyncScheduler , Observable , SchedulerLike } from 'rxjs' ; import { Action , DocumentReference , DocumentSnapshot , Query , QuerySnapshot , Reference } from '../interfaces' ; import { map , pairwise , startWith } from 'rxjs/operators' ; function _fromRef < T , R > ( ref , scheduler = asyncScheduler ) { return new Observable ( subscriber => { let unsubscribe ; if ( scheduler != null ) { scheduler . schedule ( ( ) => { unsubscribe = ref . onSnapshot ( { includeMetadataChanges : true } , subscriber ) ; } ) ; } else { unsubscribe = ref . onSnapshot ( { includeMetadataChanges : true } , subscriber ) ; } return ( ) => { if ( unsubscribe != null ) { unsubscribe ( ) ; } } ; } ) ; } export function fromRef < R , T > ( ref , scheduler ? ) { return _fromRef < typeof ref , R > ( ref , scheduler ) ; } export function fromDocRef < T > ( ref , scheduler ? ) { return fromRef < DocumentSnapshot < T > , T > ( ref , scheduler ) . pipe ( startWith < DocumentSnapshot < T > , undefined > ( undefined ) , pairwise ( ) , map ( ( [ priorPayload , payload ] ) => { if ( ! payload . exists ) { return { payload , type : 'removed' } ; } if ( ! priorPayload ?. exists ) { return { payload , type : 'added' } ; } return { payload , type : 'modified' } ; } ) ) ; } export function fromCollectionRef < T > ( ref , scheduler ? ) { return fromRef < QuerySnapshot < T > , T > ( ref , scheduler ) . pipe ( map ( payload => ( { payload , type : 'query' } ) ) ) ; }