import * as chaiAsPromised from "chai-as-promised" ; import * as chai from "chai" ; chai . use ( chaiAsPromised ) ; import { endCatcher , mapper } from "../lib/transformers" ; import "./mocha-init" ; import { useFakeTimers } from "sinon" ; import Stream from "../lib" ; import { expect } from "chai" ; import { noop } from "../lib/util" ; import { identity } from "./util" ; import BaseError from "../lib/BaseError" ; describe ( "transformers" , ( ) => { let s ; let abortError ; let boomError ; let sinonClock ; before ( ( ) => { sinonClock = useFakeTimers ( ) ; } ) ; beforeEach ( ( ) => { s = new Stream < number > ( ) ; abortError = new Error ( "Test stream explicitly aborted" ) ; boomError = new Error ( "Test error" ) ; } ) ; after ( ( ) => { sinonClock . restore ( ) ; } ) ; param fn function clockwise ( fn ) { return async ( ) => { const fnPromise = fn ( ) ; await Promise . all ( [ sinonClock . runAllAsync ( ) , fnPromise ] ) ; } ; } describe ( "endCatcher()" , ( ) => { type MyObject = number ; let expectErrorFromEnd = false ; function getRowsFromDatabase < T > ( query ) { const source = new Stream < number > ( ) ; source . write ( 1 ) ; source . write ( 2 ) ; if ( expectErrorFromEnd ) { expect ( source . end ( ) ) . eventually . rejectedWith ( NotAllObjectsProcessedError ) ; } else { expect ( source . end ( ) ) . eventually . equal ( undefined ) ; } return source ; } const process = noop ; const someCondition = true ; const myQuery = "" ; class NotAllObjectsProcessedError extends BaseError { public constructor ( ) { super ( "NotAllObjectsProcessedError" , "Not all objects were successfully processed" ) ; } } describe ( "docs example" , ( ) => { it ( "passes part 1 of example from docs" , clockwise ( async ( ) => { expectErrorFromEnd = true ; const resultPromise = getRowsFromDatabase < MyObject > ( myQuery ) . transform ( mapper ( identity ) ) . forEach ( process , ( ) => { if ( someCondition ) { throw new NotAllObjectsProcessedError ( ) ; } } ) ; await expect ( resultPromise ) . eventually . rejected . and . be . instanceOf ( NotAllObjectsProcessedError ) ; } ) ) ; it ( "passes part 2 of example from docs" , clockwise ( async ( ) => { expectErrorFromEnd = false ; const resultPromise = getRowsFromDatabase < MyObject > ( myQuery ) . transform ( endCatcher ( ( err ) => { if ( err instanceof NotAllObjectsProcessedError ) { } else { throw err ; } } ) ) . transform ( mapper ( identity ) ) . forEach ( process , ( ) => { if ( someCondition ) { throw new NotAllObjectsProcessedError ( ) ; } } ) ; await expect ( resultPromise ) . eventually . equals ( undefined ) ; } ) ) ; } ) ; it ( "Indeed throws an error if endCatcher() throws" , clockwise ( async ( ) => { function getSource ( ) { const source = new Stream < number > ( ) ; source . write ( 1 ) ; source . write ( 2 ) ; expect ( source . end ( ) ) . to . eventually . rejectedWith ( Error ) ; return source ; } const resultPromise = getSource ( ) . transform ( endCatcher ( ( ) => { throw new Error ( "B" ) ; } ) ) . forEach ( noop , ( ) => { throw new Error ( "A" ) ; } ) ; await expect ( resultPromise ) . eventually . rejectedWith ( "B" ) ; } ) ) ; } ) ; } ) ;