import { BehaviorSubject , isObservable , Observable , Subject , Subscriber , Subscription , TeardownLogic } from 'rxjs' ; import { arrayRemoveItem , createStack , isFunction , mergePromiseStack , mergeStack } from '@deepkit/core' ; import { first , skip } from 'rxjs/operators' ; export class AsyncSubscription { protected unsubscribed = false ; constructor ( private cb ) { } async unsubscribe ( ) { if ( this . unsubscribed ) return ; this . unsubscribed = true ; await this . cb ( ) ; } } export function isSubject ( v ) { return v && isFunction ( v . unsubscribe ) && isObservable ( v ) ; } export function isBehaviorSubject ( v ) { return v && isFunction ( v . getValue ) && isSubject ( v ) ; } example export class Subscriptions { public readonly list : Subscription [ ] = [ ] ; constructor ( protected teardown ? ) { } public set add ( v ) { this . list . push ( v ) ; v . add ( ( ) => { arrayRemoveItem ( this . list , v ) ; } ) ; } public unsubscribe ( ) { for ( const sub of this . list . slice ( 0 ) ) { sub . unsubscribe ( ) ; } if ( this . teardown ) { this . teardown ( ) ; } this . list . splice ( 0 , this . list . length ) ; } } export function subscriptionToPromise < T > ( subscription ) { return new Promise ( ( resolve ) => { const sub = subscription . add ( ( ) => { resolve ( ) ; sub . unsubscribe ( ) ; } ) ; } ) ; } export function nextValue < T > ( o ) { if ( isFunction ( ( o as any ) . getValue ) ) { return o . pipe ( skip ( 1 ) ) . pipe ( first ( ) ) . toPromise ( ) ; } return o . pipe ( first ( ) ) . toPromise ( ) ; } export function observableToPromise < T > ( o , next ? ) { const stack = createStack ( ) ; return new Promise ( ( resolve , reject ) => { let last ; o . subscribe ( ( data ) => { if ( next ) { next ( data ) ; } last = data ; } , ( error ) => { mergeStack ( error , stack ) ; reject ( error ) ; } , ( ) => { resolve ( last ) ; } ) ; } ) ; } export function promiseToObservable < T > ( o ) { const stack = createStack ( ) ; return new Observable ( ( observer ) => { try { mergePromiseStack ( o ( ) , stack ) . then ( ( data ) => { observer . next ( data ) ; observer . complete ( ) ; } , ( error ) => { observer . error ( error ) ; } ) ; } catch ( error ) { observer . error ( error ) ; } } ) ; } export async function tearDown ( teardown ) { if ( 'function' === typeof teardown ) { await teardown ( ) ; } else if ( 'object' === typeof teardown && teardown . unsubscribe ) { await teardown . unsubscribe ( ) ; } }