import { Observable } from "rxjs" ; import { Deferred } from "./util/deferred" ; const RESOLVED = Promise . resolve ( ) ; param source export async function * eachValueFrom < T > ( source ) { const deferreds = [ ] ; const values = [ ] ; let hasError = false ; let error = null ; let completed = false ; const subs = source . subscribe ( { next : value => { if ( deferreds . length > 0 ) { deferreds . shift ( ) ! . resolve ( { value , done : false } ) ; } else { values . push ( value ) ; } } , error : err => { hasError = true ; error = err ; while ( deferreds . length > 0 ) { deferreds . shift ( ) ! . reject ( err ) ; } } , complete : ( ) => { completed = true ; while ( deferreds . length > 0 ) { deferreds . shift ( ) ! . resolve ( { value : undefined , done : true } ) ; } } , } ) ; try { while ( true ) { if ( values . length > 0 ) { yield values . shift ( ) ! ; } else if ( completed ) { return ; } else if ( hasError ) { throw error ; } else { const d = new Deferred < IteratorResult < T > > ( ) ; deferreds . push ( d ) ; const result = await d . promise ; if ( result . done ) { return ; } else { yield result . value ; } } } } catch ( err ) { throw err ; } finally { subs . unsubscribe ( ) ; } } param source export async function * bufferedValuesFrom < T > ( source ) { let deferred = null ; const buffer = [ ] ; let hasError = false ; let error = null ; let completed = false ; const subs = source . subscribe ( { next : value => { if ( deferred ) { deferred . resolve ( RESOLVED . then ( ( ) => { const bufferCopy = buffer . slice ( ) ; buffer . length = 0 ; return { value : bufferCopy , done : false } ; } ) ) ; deferred = null ; } buffer . push ( value ) ; } , error : err => { hasError = true ; error = err ; if ( deferred ) { deferred . reject ( err ) ; deferred = null ; } } , complete : ( ) => { completed = true ; if ( deferred ) { deferred . resolve ( { value : undefined , done : true } ) ; deferred = null ; } } , } ) ; try { while ( true ) { if ( buffer . length > 0 ) { const bufferCopy = buffer . slice ( ) ; buffer . length = 0 ; yield bufferCopy ; } else if ( completed ) { return ; } else if ( hasError ) { throw error ; } else { deferred = new Deferred < IteratorResult < T [ ] > > ( ) ; const result = await deferred . promise ; if ( result . done ) { return ; } else { yield result . value ; } } } } catch ( err ) { throw err ; } finally { subs . unsubscribe ( ) ; } } param source export async function * latestValueFrom < T > ( source ) { let deferred = undefined ; let latestValue ; let hasLatestValue = false ; let hasError = false ; let error = null ; let completed = false ; const subs = source . subscribe ( { next : value => { hasLatestValue = true ; latestValue = value ; if ( deferred ) { deferred . resolve ( RESOLVED . then ( ( ) => { hasLatestValue = false ; return { value : latestValue , done : false } ; } ) ) ; } } , error : err => { hasError = true ; error = err ; if ( deferred ) { deferred . reject ( err ) ; } } , complete : ( ) => { completed = true ; if ( deferred ) { hasLatestValue = false ; deferred . resolve ( { value : undefined , done : true } ) ; } } , } ) ; try { while ( true ) { if ( hasLatestValue ) { await RESOLVED ; const value = latestValue ! ; hasLatestValue = false ; yield value ; } else if ( completed ) { return ; } else if ( hasError ) { throw error ; } else { deferred = new Deferred < IteratorResult < T > > ( ) ; const result = await deferred . promise ; if ( result . done ) { return ; } else { yield result . value ; } } } } catch ( err ) { throw err ; } finally { subs . unsubscribe ( ) ; } } param source export async function * nextValueFrom < T > ( source ) { let deferred = undefined ; let hasError = false ; let error = null ; let completed = false ; const subs = source . subscribe ( { next : value => { if ( deferred ) { deferred . resolve ( { value , done : false } ) ; } } , error : err => { hasError = true ; error = err ; if ( deferred ) { deferred . reject ( err ) ; } } , complete : ( ) => { completed = true ; if ( deferred ) { deferred . resolve ( { value : undefined , done : true } ) ; } } , } ) ; try { while ( true ) { if ( completed ) { return ; } else if ( hasError ) { throw error ; } else { deferred = new Deferred < IteratorResult < T > > ( ) ; const result = await deferred . promise ; if ( result . done ) { return ; } else { yield result . value ; } } } } catch ( err ) { throw err ; } finally { subs . unsubscribe ( ) ; } }