import { expect } from 'chai' ; import { concat , defer , Observable , of , throwError , EMPTY , from } from 'rxjs' ; import { catchError , map , mergeMap , takeWhile , delay , take } from 'rxjs/operators' ; import * as sinon from 'sinon' ; import { createObservableInputs } from '../helpers/test-helper' ; import { TestScheduler } from 'rxjs/testing' ; import { observableMatcher } from '../helpers/observableMatcher' ; import { asInteropObservable } from '../helpers/interop-helper' ; test describe ( 'catchError operator' , ( ) => { let testScheduler ; beforeEach ( ( ) => { testScheduler = new TestScheduler ( observableMatcher ) ; } ) ; it ( 'should catch error and replace with a cold Observable' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable } ) => { const e1 = hot ( '  --a--b--#       ' ) ; const e2 = cold ( '         -1-2-3-|' ) ; const expected = '--a--b---1-2-3-|' ; const result = e1 . pipe ( catchError ( ( err ) => e2 ) ) ; expectObservable ( result ) . toBe ( expected ) ; } ) ; } ) ; it ( 'should catch error and replace it with Observable.of()' , ( ) => { testScheduler . run ( ( { hot , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--c--------|' ) ; const subs = '    ^-------!' ; const expected = '--a--b--(XYZ|)' ; const result = e1 . pipe ( map ( ( n ) => { if ( n === 'c' ) { throw 'bad' ; } return n ; } ) , catchError ( ( err ) => { return of ( 'X' , 'Y' , 'Z' ) ; } ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( subs ) ; } ) ; } ) ; it ( 'should catch error and replace it with a cold Observable' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--#          ' ) ; const e1subs = '  ^-------!          ' ; const e2 = cold ( '         1-2-3-4-5-|' ) ; const e2subs = '  --------^---------!' ; const expected = '--a--b--1-2-3-4-5-|' ; const result = e1 . pipe ( catchError ( ( err ) => e2 ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should allow unsubscribing explicitly and early' , ( ) => { testScheduler . run ( ( { hot , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --1-2-3-4-5-6---#' ) ; const e1subs = '  ^------!         ' ; const expected = '--1-2-3-         ' ; const unsub = '   -------!         ' ; const result = e1 . pipe ( catchError ( ( ) => { return of ( 'X' , 'Y' , 'Z' ) ; } ) ) ; expectObservable ( result , unsub ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; } ) ; } ) ; it ( 'should not break unsubscription chain when unsubscribed explicitly' , ( ) => { testScheduler . run ( ( { hot , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --1-2-3-4-5-6---#' ) ; const e1subs = '  ^------!         ' ; const expected = '--1-2-3-         ' ; const unsub = '   -------!         ' ; const result = e1 . pipe ( mergeMap ( ( x ) => of ( x ) ) , catchError ( ( ) => { return of ( 'X' , 'Y' , 'Z' ) ; } ) , mergeMap ( ( x ) => of ( x ) ) ) ; expectObservable ( result , unsub ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; } ) ; } ) ; it ( 'should unsubscribe from a caught hot caught observable when unsubscribed explicitly' , ( ) => { testScheduler . run ( ( { hot , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  -1-2-3-#          ' ) ; const e1subs = '  ^------!          ' ; const e2 = hot ( '  ---3-4-5-6-7-8-9-|' ) ; const e2subs = '  -------^----!     ' ; const expected = '-1-2-3-5-6-7-     ' ; const unsub = '   ------------!     ' ; const result = e1 . pipe ( catchError ( ( ) => e2 ) ) ; expectObservable ( result , unsub ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should unsubscribe from a caught cold caught observable when unsubscribed explicitly' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  -1-2-3-#          ' ) ; const e1subs = '  ^------!          ' ; const e2 = cold ( '       5-6-7-8-9-|' ) ; const e2subs = '  -------^----!     ' ; const expected = '-1-2-3-5-6-7-     ' ; const unsub = '   ------------!     ' ; const result = e1 . pipe ( catchError ( ( ) => e2 ) ) ; expectObservable ( result , unsub ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should unsubscribe from a caught cold caught interop observable when unsubscribed explicitly' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  -1-2-3-#          ' ) ; const e1subs = '  ^------!          ' ; const e2 = cold ( '       5-6-7-8-9-|' ) ; const e2subs = '  -------^----!     ' ; const expected = '-1-2-3-5-6-7-     ' ; const unsub = '   ------------!     ' ; const result = e1 . pipe ( catchError ( ( ) => asInteropObservable ( e2 ) ) ) ; expectObservable ( result , unsub ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should stop listening to a synchronous observable when unsubscribed' , ( ) => { const sideEffects = [ ] ; const synchronousObservable = concat ( defer ( ( ) => { sideEffects . push ( 1 ) ; return of ( 1 ) ; } ) , defer ( ( ) => { sideEffects . push ( 2 ) ; return of ( 2 ) ; } ) , defer ( ( ) => { sideEffects . push ( 3 ) ; return of ( 3 ) ; } ) ) ; throwError ( ( ) => new Error ( 'Some error' ) ) . pipe ( catchError ( ( ) => synchronousObservable ) , takeWhile ( ( x ) => x != 2 ) ) . subscribe ( ( ) => { } ) ; expect ( sideEffects ) . to . deep . equal ( [ 1 , 2 ] ) ; } ) ; it ( 'should catch error and replace it with a hot Observable' , ( ) => { testScheduler . run ( ( { hot , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--#          ' ) ; const e1subs = '  ^-------!          ' ; const e2 = hot ( '  1-2-3-4-5-6-7-8-9-|' ) ; const e2subs = '  --------^---------!' ; const expected = '--a--b--5-6-7-8-9-|' ; const result = e1 . pipe ( catchError ( ( err ) => e2 ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should catch and allow the cold observable to be repeated with the third ' + '(caught) argument' , ( ) => { testScheduler . run ( ( { cold , expectObservable , expectSubscriptions } ) => { const e1 = cold ( '--a--b--c--------|       ' ) ; const subs = [ '               ^-------!                ' , '              --------^-------!         ' , '              ----------------^-------! ' ] ; const expected = '--a--b----a--b----a--b--#' ; let retries = 0 ; const result = e1 . pipe ( map ( ( n ) => { if ( n === 'c' ) { throw 'bad' ; } return n ; } ) , catchError ( ( err , caught ) => { if ( retries ++ === 2 ) { throw 'done' ; } return caught ; } ) ) ; expectObservable ( result ) . toBe ( expected , undefined , 'done' ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( subs ) ; } ) ; } ) ; it ( 'should catch and allow the hot observable to proceed with the third ' + '(caught) argument' , ( ) => { testScheduler . run ( ( { hot , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--c----d---|' ) ; const subs = [ '               ^-------!         ' , '              --------^--------! ' ] ; const expected = '--a--b-------d---|' ; let retries = 0 ; const result = e1 . pipe ( map ( ( n ) => { if ( n === 'c' ) { throw 'bad' ; } return n ; } ) , catchError ( ( err , caught ) => { if ( retries ++ === 2 ) { throw 'done' ; } return caught ; } ) ) ; expectObservable ( result ) . toBe ( expected , undefined , 'done' ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( subs ) ; } ) ; } ) ; it ( 'should catch and replace a Observable.throw() as the source' , ( ) => { testScheduler . run ( ( { cold , expectObservable , expectSubscriptions } ) => { const e1 = cold ( ' #' ) ; const subs = '    (^!)' ; const expected = '(abc|)' ; const result = e1 . pipe ( catchError ( ( err ) => of ( 'a' , 'b' , 'c' ) ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( subs ) ; } ) ; } ) ; it ( 'should mirror the source if it does not raise errors' , ( ) => { testScheduler . run ( ( { cold , expectObservable , expectSubscriptions } ) => { const e1 = cold ( ' --a--b--c--|' ) ; const subs = '    ^----------!' ; const expected = '--a--b--c--|' ; const result = e1 . pipe ( catchError ( ( err ) => of ( 'x' , 'y' , 'z' ) ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( subs ) ; } ) ; } ) ; it ( 'should complete if you return Observable.empty()' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--#' ) ; const e1subs = '  ^-------!' ; const e2 = cold ( '         |' ) ; const e2subs = '  --------(^!)' ; const expected = '--a--b--|' ; const result = e1 . pipe ( catchError ( ( ) => e2 ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should raise error if you return Observable.throw()' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--#' ) ; const e1subs = '  ^-------!' ; const e2 = cold ( '         #' ) ; const e2subs = '  --------(^!)' ; const expected = '--a--b--#' ; const result = e1 . pipe ( catchError ( ( ) => e2 ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should never terminate if you return NEVER' , ( ) => { testScheduler . run ( ( { hot , cold , expectObservable , expectSubscriptions } ) => { const e1 = hot ( '  --a--b--#' ) ; const e1subs = '  ^-------!' ; const e2 = cold ( '         -' ) ; const e2subs = '  --------^' ; const expected = '--a--b---' ; const result = e1 . pipe ( catchError ( ( ) => e2 ) ) ; expectObservable ( result ) . toBe ( expected ) ; expectSubscriptions ( e1 . subscriptions ) . toBe ( e1subs ) ; expectSubscriptions ( e2 . subscriptions ) . toBe ( e2subs ) ; } ) ; } ) ; it ( 'should pass the error as the first argument' , ( done ) => { throwError ( ( ) => ( 'bad' ) ) . pipe ( catchError ( ( err ) => { expect ( err ) . to . equal ( 'bad' ) ; return EMPTY ; } ) ) . subscribe ( { next : ( ) => { } , error : ( err ) => { done ( new Error ( 'should not be called' ) ) ; } , complete : ( ) => { done ( ) ; } } ) ; } ) ; it ( 'should accept selector returns any ObservableInput' , ( done ) => { const input$ = createObservableInputs ( 42 ) ; input$ . pipe ( mergeMap ( input => throwError ( ( ) => ( 'bad' ) ) . pipe ( catchError ( err => input ) ) ) ) . subscribe ( { next : x => { expect ( x ) . to . be . equal ( 42 ) ; } , error : ( err ) => { done ( new Error ( 'should not be called' ) ) ; } , complete : ( ) => { done ( ) ; } } ) ; } ) ; it ( 'should catch errors throw from within the constructor' , ( ) => { testScheduler . run ( ( { expectObservable } ) => { const source = concat ( new Observable < string > ( o => { o . next ( 'a' ) ; throw 'kaboom' ; } ) . pipe ( catchError ( _ => of ( 'b' ) ) ) , of ( 'c' ) ) ; const expected = '(abc|)' ; expectObservable ( source ) . toBe ( expected ) ; } ) ; } ) ; context ( 'fromPromise' , ( ) => { type SetTimeout = ( callback : ( ... args : any [ ] ) => void , ms : number , ... args : any [ ] ) => NodeJS . Timer ; let trueSetTimeout ; let sandbox ; let timers ; beforeEach ( ( ) => { trueSetTimeout = global . setTimeout ; sandbox = sinon . createSandbox ( ) ; timers = sandbox . useFakeTimers ( ) ; } ) ; afterEach ( ( ) => { sandbox . restore ( ) ; } ) ; it ( 'should chain a throw from a promise using Observable.throw' , ( done ) => { const subscribeSpy = sinon . spy ( ) ; const errorSpy = sinon . spy ( ) ; const thrownError = new Error ( 'BROKEN THROW' ) ; const testError = new Error ( 'BROKEN PROMISE' ) ; from ( Promise . reject ( testError ) ) . pipe ( catchError ( err => throwError ( ( ) => ( thrownError ) ) ) ) . subscribe ( { next : subscribeSpy , error : errorSpy } ) ; trueSetTimeout ( ( ) => { try { timers . tick ( 1 ) ; } catch ( e ) { return done ( new Error ( 'This should not have thrown an error' ) ) ; } expect ( subscribeSpy ) . not . to . be . called ; expect ( errorSpy ) . to . have . been . called ; expect ( errorSpy ) . to . have . been . calledWith ( thrownError ) ; done ( ) ; } , 0 ) ; } ) ; } ) ; it ( 'Properly handle async handled result if source is synchronous' , ( done ) => { const source = new Observable < string > ( observer => { observer . error ( new Error ( 'kaboom!' ) ) ; observer . complete ( ) ; } ) ; const sourceWithDelay = new Observable < string > ( observer => { observer . next ( 'delayed' ) ; observer . complete ( ) ; } ) . pipe ( delay ( 0 ) ) ; const values = [ ] ; source . pipe ( catchError ( err => sourceWithDelay ) ) . subscribe ( { next : value => values . push ( value ) , error : err => done ( err ) , complete : ( ) => { expect ( values ) . to . deep . equal ( [ 'delayed' ] ) ; done ( ) ; } } ) ; } ) ; it ( 'should stop listening to a synchronous observable when unsubscribed' , ( ) => { const sideEffects = [ ] ; const synchronousObservable = new Observable ( subscriber => { for ( let i = 0 ; ! subscriber . closed && i < 10 ; i ++ ) { sideEffects . push ( i ) ; subscriber . next ( i ) ; } } ) ; synchronousObservable . pipe ( catchError ( ( ) => EMPTY ) , take ( 3 ) ) . subscribe ( ( ) => { } ) ; expect ( sideEffects ) . to . deep . equal ( [ 0 , 1 , 2 ] ) ; } ) ; } ) ;