/**
 * Promise-based object stream with seamless support for back-pressure and error
 * handling, written in Typescript.
 *
 * Copyright (C) 2015 Martin Poelstra
 * License: MIT
 */ import { Readable , Writable } from "./Stream" ; import { batch , filter , map , Transform } from "./Transform" ; export function mapper < In , Out > ( mapFn ) { return ( readable , writable ) : void => { map ( readable , writable , mapFn ) ; } ; } export function filterer < T > ( filterFn ) { return ( readable , writable ) : void => { filter ( readable , writable , filterFn ) ; } ; } export type BatcherOptions < T > = { minBatchSize ? : number ; flushTimeout ? : number ; handleError ? : ( e : Error , batch : T [ ] ) => void | PromiseLike < void > ; } ; param maxBatchSize param options . minBatchSize param options . flushTimeout param options . handleError return export function batcher < T > ( maxBatchSize , options = { } ) { return ( readable , writable ) : void => { batch ( readable , writable , maxBatchSize , options ) ; } ; } example param handleError export function endCatcher < T > ( handleError ) { return ( readable , writable ) : void => { writable . aborted ( ) . catch ( ( err ) => readable . abort ( err ) ) ; readable . aborted ( ) . catch ( ( err ) => writable . abort ( err ) ) ; readable . forEach ( ( value ) => writable . write ( value ) , ( error ? ) => writable . end ( error , readable . result ( ) ) . catch ( handleError ) ) ; } ; }