import '../asynciterablehelpers' ; import Ix from 'ix/Ix' ; import { empty , from , of , throwError , toArray , toObservable } from 'ix/asynciterable' ; import { Observable as RxJSObservable , from as RxJSObservableFrom } from 'rxjs' ; import { Observable , PartialObserver } from '../../src/observer' ; test ( 'AsyncIterable#toObservable empty' , async ( ) => { const xs = empty ( ) ; const ys = toObservable ( xs ) ; let fail = false ; ys . subscribe ( { next : ( _value ) => { fail = true ; } , error : ( _err ) => { fail = true ; } , complete : ( ) => { expect ( fail ) . toBeFalsy ( ) ; } , } ) ; } ) ; test ( 'AsyncIterable#toObservable non-empty' , async ( ) => { const results = [ ] ; const xs = of ( 1 , 2 , 3 ) ; const ys = toObservable ( xs ) ; let fail = false ; ys . subscribe ( { next : ( value ) => { results . push ( value ) ; } , error : ( _err ) => { fail = true ; } , complete : ( ) => { expect ( results ) . toEqual ( [ 1 , 2 , 3 ] ) ; expect ( fail ) . toBeFalsy ( ) ; } , } ) ; } ) ; test ( 'AsyncIterable#toObservable error' , async ( ) => { const error = new Error ( ) ; const xs = throwError ( error ) ; const ys = toObservable ( xs ) ; let fail = false ; ys . subscribe ( { next : ( _value ) => { fail = true ; } , error : ( err ) => { expect ( err ) . toEqual ( error ) ; expect ( fail ) . toBeFalsy ( ) ; } , complete : ( ) => { fail = true ; } , } ) ; } ) ; test ( 'AsyncIterable#toObservable Symbol.observable should return same instance' , async ( ) => { const ys = toObservable ( of ( 1 , 2 , 3 ) ) ; expect ( ys ) . toBe ( ys [ Ix . symbolObservable ] ( ) ) ; } ) ; test ( 'AsyncIterable#toObservable accepts partial observers' , async ( ) => { const expectedValues = [ 1 , 2 , 3 ] ; const expectedError = new Error ( ) ; const actualValues = [ ] ; let actualError = null ; let completeCalled = false ; const valueObs = toObservable ( of ( 1 , 2 , 3 ) ) ; const errorObs = toObservable ( throwError ( expectedError ) ) ; const onNext = ( val ) => actualValues . push ( val ) ; const onError = ( error ) => ( actualError = error ) ; const onComplete = ( ) => ( completeCalled = true ) ; await endOfObservable ( valueObs , { next : onNext } ) ; try { await endOfObservable ( errorObs , { error : onError } ) ; } catch ( e ) { } await endOfObservable ( valueObs , { complete : onComplete } ) ; expect ( actualValues ) . toEqual ( expectedValues ) ; expect ( actualError ) . toEqual ( expectedError ) ; expect ( completeCalled ) . toEqual ( true ) ; } ) ; test ( 'AsyncIterable#toObservable accepts observer functions' , async ( ) => { const expectedValues = [ 1 , 2 , 3 ] ; const expectedError = new Error ( ) ; const actualValues = [ ] ; let actualError = null ; let completeCalled = false ; const valueObs = toObservable ( of ( 1 , 2 , 3 ) ) ; const errorObs = toObservable ( throwError ( expectedError ) ) ; const onNext = ( val ) => actualValues . push ( val ) ; const onError = ( error ) => ( actualError = error ) ; const onComplete = ( ) => ( completeCalled = true ) ; await endOfObservable ( valueObs , onNext ) ; try { await endOfObservable ( errorObs , null , onError ) ; } catch ( e ) { } await endOfObservable ( valueObs , null , null , onComplete ) ; expect ( actualValues ) . toEqual ( expectedValues ) ; expect ( actualError ) . toEqual ( expectedError ) ; expect ( completeCalled ) . toEqual ( true ) ; } ) ; test ( 'AsyncIterable#toObservable interop with rxjs' , async ( ) => { const xs = [ ] ; const ys = RxJSObservableFrom ( toObservable ( of ( 1 , 2 , 3 ) ) ) ; await endOfObservable ( ys , ( x ) => xs . push ( x ) ) ; expect ( xs ) . toEqual ( [ 1 , 2 , 3 ] ) ; } ) ; test ( 'AsyncIterable.from interop with rxjs' , async ( ) => { const ys = from ( RxJSObservableFrom ( toObservable ( of ( 1 , 2 , 3 ) ) ) ) ; const xs = await toArray ( ys ) ; expect ( xs ) . toEqual ( [ 1 , 2 , 3 ] ) ; } ) ; function endOfObservable < T > ( observable , next ? , error ? , complete ? ) { let reject ; let resolve ; const done = new Promise < void > ( ( a , b ) => { resolve = a ; reject = b ; } ) ; const wrap = ( promiseFn , originalFn ) => ( ... args ) => { promiseFn ( ... args ) ; originalFn ( ... args ) ; } ; if ( next && typeof next === 'object' ) { next . error = wrap ( e => reject ( e ) , ( next . error || ( ( ) => { } ) ) . bind ( next ) ) ; next . complete = wrap ( ( ) => resolve ( ) , ( next . complete || ( ( ) => { } ) ) . bind ( next ) ) ; } else { error = wrap ( e => reject ( e ) , error || ( ( ) => { } ) ) ; complete = wrap ( ( ) => resolve ( ) , complete || ( ( ) => { } ) ) ; } observable . subscribe ( < any > next , < any > error , < any > complete ) ; return done ; }